基因编辑与生物信息安全取证深度分析

#!/bin/bash
BAM_FILE=$1
EXPECTED_MD5=$2

ACTUAL_MD5=$(md5sum "$BAM_FILE" | awk '{print $1}')

if [ "$ACTUAL_MD5" != "$EXPECTED_MD5" ]; then
    echo "[ALERT] MD5 校验失败: $BAM_FILE"
    echo "  期望: $EXPECTED_MD5"
    echo "  实际: $ACTUAL_MD5"
    echo "[FORENSIC] 检查文件元数据..."
    stat "$BAM_FILE"
    samtools view -H "$BAM_FILE" | head -20
    exit 1
fi

echo "[OK] MD5 校验通过: $BAM_FILE"

BAM_HEADER=$(samtools view -H "$BAM_FILE" | grep "^@SQ" | wc -l)
BAM_READS=$(samtools idxstats "$BAM_FILE" | awk '{s+=$3} END {print s}')
echo "[INFO] 参考序列条目数: $BAM_HEADER"
echo "[INFO] 比对读段总数: $BAM_READS"

samtools flagstat "$BAM_FILE"

echo "[CHECK] 验证 BAM 索引一致性..."
samtools quickcheck -vv "$BAM_FILE" 2>&1

import pandas as pd
import numpy as np
from collections import Counter
import sys
import json

def load_sgrna_designs(file_path):
    designs = []
    with open(file_path, 'r') as f:
        for line in f:
            if line.startswith('#') or line.strip() == '':
                continue
            fields = line.strip().split('\t')
            if len(fields) >= 6:
                designs.append({
                    'sgrna_id': fields[0],
                    'target_gene': fields[1],
                    'sequence': fields[2],
                    'on_target_score': float(fields[3]),
                    'off_target_score': float(fields[4]),
                    'gc_content': float(fields[5])
                })
    return pd.DataFrame(designs)

def detect_sequence_anomalies(df, gc_bounds=(0.35, 0.70), on_target_min=0.5):
    anomalies = []
    for idx, row in df.iterrows():
        seq = row['sequence']
        if len(seq) < 17 or len(seq) > 23:
            anomalies.append({
                'row': idx, 'type': 'LENGTH_ANOMALY',
                'detail': f"sgRNA长度异常: {len(seq)}nt",
                'severity': 'HIGH'
            })
        gc = sum(1 for c in seq if c in 'GC') / len(seq) if len(seq) > 0 else 0
        if gc < gc_bounds[0] or gc > gc_bounds[1]:
            anomalies.append({
                'row': idx, 'type': 'GC_CONTENT_ANOMALY',
                'detail': f"GC含量异常: {gc:.2%}",
                'severity': 'MEDIUM'
            })
        if 'TTTT' in seq:
            anomalies.append({
                'row': idx, 'type': 'POLY_T_MOTIF',
                'detail': '检测到连续4个T，可能触发U6终止子',
                'severity': 'HIGH'
            })
        if row['on_target_score'] < on_target_min:
            anomalies.append({
                'row': idx, 'type': 'LOW_ON_TARGET',
                'detail': f"On-target评分过低: {row['on_target_score']}",
                'severity': 'MEDIUM'
            })
    return anomalies

def detect_off_target_tampering(df, baseline_off_target_mean=0.3):
    off_target_scores = df['off_target_score'].values
    mean_val = np.mean(off_target_scores)
    std_val = np.std(off_target_scores)
    tampering_suspect = []
    if abs(mean_val - baseline_off_target_mean) > 0.15:
        tampering_suspect.append({
            'type': 'DISTRIBUTION_SHIFT',
            'detail': f"Off-target评分分布偏移: 均值{mean_val:.3f} vs 基线{baseline_off_target_mean}",
            'severity': 'CRITICAL'
        })
    cv = std_val / mean_val if mean_val != 0 else float('inf')
    if cv < 0.05:
        tampering_suspect.append({
            'type': 'SUSPICIOUS_UNIFORMITY',
            'detail': f"Off-target评分异常均匀(CV={cv:.4f})，疑似伪造",
            'severity': 'HIGH'
        })
    return tampering_suspect

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("用法: python sgrna_audit.py <sgrna_designs.tsv>")
        sys.exit(1)
    df = load_sgrna_designs(sys.argv[1])
    print(f"[INFO] 加载 {len(df)} 条 sgRNA 设计记录")
    seq_anomalies = detect_sequence_anomalies(df)
    off_target_flags = detect_off_target_tampering(df)
    report = {
        'total_sgrnas': len(df),
        'sequence_anomalies': seq_anomalies,
        'off_target_tampering': off_target_flags,
        'risk_level': 'CRITICAL' if any(a['severity'] == 'CRITICAL' for a in seq_anomalies + off_target_flags) else 'MEDIUM' if seq_anomalies else 'LOW'
    }
    print(json.dumps(report, indent=2, ensure_ascii=False))
    if report['risk_level'] == 'CRITICAL':
        print("[ALERT] 检测到严重异常，建议立即暂停实验并进行深度取证分析")

#!/bin/bash
echo "=== 测序仪网络连接审计 ==="

echo "[1] 检查测序仪IP段的网络连接..."
SEQUENCER_SUBNET="10.20.30.0/24"
netstat -an | grep "$SEQUENCER_SUBNET" | awk '{print $5}' | sort -u > /tmp/sequencer_connections.txt
cat /tmp/sequencer_connections.txt

echo "[2] 检查到外部网络的连接（可能的数据外泄）..."
EXTERNAL_CONNS=$(netstat -an | grep "ESTABLISHED" | grep -v "10.0.0.0\|172.16.0.0\|192.168.0.0\|127.0.0.1" | wc -l)
echo "  外部连接数: $EXTERNAL_CONNS"
if [ "$EXTERNAL_CONNS" -gt 5 ]; then
    echo "  [ALERT] 发现大量外部连接，可能存在数据外泄"
    netstat -an | grep "ESTABLISHED" | grep -v "10.0.0.0\|172.16.0.0\|192.168.0.0\|127.0.0.1"
fi

echo "[3] 检查DNS查询日志..."
if [ -f /var/log/dnsmasq.log ]; then
    grep "sequencer\|illumina\|nanopore\|pacbio" /var/log/dnsmasq.log | tail -20
fi

echo "[4] 检查测序仪固件版本..."
curl -s "http://10.20.30.100/api/system/version" 2>/dev/null || echo "  无法访问测序仪管理接口"

echo "[5] 检查异常进程..."
ps aux | grep -i "bcl2fastq\|bcl-convert\|guppy\|basecaller" | grep -v grep

import hashlib
import os
import json
import subprocess
from datetime import datetime
from pathlib import Path

def compute_file_hash(file_path, algorithm='sha256'):
    hash_func = hashlib.new(algorithm)
    with open(file_path, 'rb') as f:
        for chunk in iter(lambda: f.read(8192), b''):
            hash_func.update(chunk)
    return hash_func.hexdigest()

def audit_bam_file(bam_path):
    result = {}
    result['file_path'] = bam_path
    result['file_size'] = os.path.getsize(bam_path)
    result['sha256'] = compute_file_hash(bam_path)
    result['mtime'] = datetime.fromtimestamp(os.path.getmtime(bam_path)).isoformat()
    result['ctime'] = datetime.fromtimestamp(os.path.getctime(bam_path)).isoformat()
    try:
        header = subprocess.run(
            ['samtools', 'view', '-H', bam_path],
            capture_output=True, text=True, timeout=30
        )
        result['header_lines'] = header.stdout.strip().split('\n')[:10]
        result['has_pg_tag'] = any('@PG' in line for line in result['header_lines'])
    except Exception as e:
        result['header_error'] = str(e)
    try:
        stats = subprocess.run(
            ['samtools', 'idxstats', bam_path],
            capture_output=True, text=True, timeout=60
        )
        total_reads = sum(
            int(line.split('\t')[2])
            for line in stats.stdout.strip().split('\n')
            if line and '\t' in line
        )
        result['total_aligned_reads'] = total_reads
    except Exception as e:
        result['stats_error'] = str(e)
    return result

def verify_bam_index(bam_path):
    idx_path = bam_path + '.bai'
    if not os.path.exists(idx_path):
        return {'status': 'MISSING', 'message': 'BAM索引文件缺失'}
    result = {
        'status': 'PRESENT',
        'index_size': os.path.getsize(idx_path),
        'index_sha256': compute_file_hash(idx_path),
        'index_mtime': datetime.fromtimestamp(os.path.getmtime(idx_path)).isoformat()
    }
    if os.path.getmtime(idx_path) < os.path.getmtime(bam_path):
        result['WARNING'] = '索引文件比BAM文件旧，可能BAM被修改后未重新建索引'
    return result

if __name__ == '__main__':
    import sys
    if len(sys.argv) < 2:
        print("用法: python bam_audit.py <bam_file>")
        sys.exit(1)
    bam_path = sys.argv[1]
    audit = audit_bam_file(bam_path)
    index = verify_bam_index(bam_path)
    audit['index_verification'] = index
    print(json.dumps(audit, indent=2, ensure_ascii=False))

#!/bin/bash
ASPERA_LOG_DIR="/var/log/aspera"
ASPERA_SESSION_DIR="$HOME/.aspera"

echo "=== Aspera 传输审计 ==="

echo "[1] 列出最近的传输会话..."
find "$ASPERA_LOG_DIR" -name "*.log" -mtime -7 -exec ls -lh {} \;

echo "[2] 检查异常传输目标..."
grep -r "destination\|remote_host" "$ASPERA_LOG_DIR"/*.log 2>/dev/null | \
    awk '{print $NF}' | sort | uniq -c | sort -rn | head -20

echo "[3] 检查大数据量传输（可能的数据外泄）..."
grep -h "bytes_written\|total_bytes" "$ASPERA_LOG_DIR"/*.log 2>/dev/null | \
    awk '{if($1 > 10737418240) print "[ALERT] 超过10GB传输: " $0}'

echo "[4] 检查传输时间异常（非工作时间）..."
grep -hE "session_start" "$ASPERA_LOG_DIR"/*.log 2>/dev/null | while read line; do
    hour=$(echo "$line" | grep -oP '\d{2}(?=:\d{2}:\d{2})' | head -1)
    if [ "$hour" -lt 6 ] || [ "$hour" -gt 22 ]; then
        echo "  [WARNING] 非工作时间传输: $line"
    fi
done

echo "[5] 检查SSH密钥配置..."
ls -la "$ASPERA_SESSION_DIR"/etc/ 2>/dev/null
cat "$ASPERA_SESSION_DIR"/etc/asperaweb_id_dsa 2>/dev/null | head -1

#!/bin/bash
LIMS_HOST=$1
echo "=== LIMS 安全审计: $LIMS_HOST ==="

echo "[1] 检查 LIMS Web 服务器访问日志中的异常..."
if [ -f /var/log/nginx/lims_access.log ]; then
    echo "  SQL注入尝试检测:"
    grep -iE "(union|select|insert|update|delete|drop|exec|xp_cmdshell|0x[0-9a-f]+)" \
        /var/log/nginx/lims_access.log | tail -20
    echo "  路径遍历检测:"
    grep -iE "(\.\./|\.\.\\\\|%2e%2e)" \
        /var/log/nginx/lims_access.log | tail -20
    echo "  高频访问IP（可能的暴力破解）:"
    awk '{print $1}' /var/log/nginx/lims_access.log | \
        sort | uniq -c | sort -rn | head -10
fi

echo "[2] 检查 LIMS 数据库连接审计..."
if [ -f /var/log/postgresql/lims_audit.log ]; then
    grep "DDL\|TRUNCATE\|DROP\|ALTER" \
        /var/log/postgresql/lims_audit.log | tail -30
    echo "  异常大查询（可能的全表导出）:"
    grep "duration:" /var/log/postgresql/lims_audit.log | \
        awk -F'duration: ' '{if($2+0 > 60) print}' | tail -10
fi

echo "[3] 检查 LIMS API 调用异常..."
if [ -f /var/log/lims/api_audit.log ]; then
    echo "  批量样本数据导出操作:"
    grep -iE "(export|download|bulk|batch)" \
        /var/log/lims/api_audit.log | tail -20
    echo "  非授权的实验设计修改:"
    grep -iE "(PUT|POST|PATCH).*(/api/v1/designs|/api/v1/protocols)" \
        /var/log/lims/api_audit.log | grep -v "authorized_admin" | tail -20
fi

echo "[4] 检查LIMS服务器上的可疑进程..."
ssh "$LIMS_HOST" "ps aux | grep -iE 'lims|laboratory|sample' | grep -v grep"

echo "[5] 检查LIMS数据库定时备份完整性..."
BACKUP_DIR="/backup/lims"
if [ -d "$BACKUP_DIR" ]; then
    find "$BACKUP_DIR" -name "*.sql.gz" -mtime -7 -exec ls -lh {} \;
    LATEST=$(ls -t "$BACKUP_DIR"/*.sql.gz 2>/dev/null | head -1)
    if [ -n "$LATEST" ]; then
        echo "  最新备份: $LATEST"
        md5sum "$LATEST"
        gunzip -t "$LATEST" && echo "  [OK] 备份文件完整" || echo "  [ALERT] 备份文件损坏"
    fi
fi

import re
import json
from datetime import datetime, timedelta
from collections import defaultdict

def parse_api_log(log_path):
    entries = []
    pattern = re.compile(
        r'(?P<timestamp>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})'
        r'\s+(?P<method>\w+)\s+(?P<endpoint>\S+)'
        r'\s+user=(?P<user>\S+)'
        r'\s+status=(?P<status>\d+)'
        r'\s+ip=(?P<ip>\S+)'
        r'\s+bytes=(?P<bytes>\d+)'
    )
    with open(log_path, 'r') as f:
        for line in f:
            m = pattern.search(line)
            if m:
                entries.append(m.groupdict())
    return entries

def detect_api_anomalies(entries):
    anomalies = []
    user_requests = defaultdict(list)
    for e in entries:
        user_requests[e['user']].append(e)
    for user, reqs in user_requests.items():
        if len(reqs) > 1000:
            anomalies.append({
                'type': 'EXCESSIVE_REQUESTS',
                'user': user,
                'count': len(reqs),
                'severity': 'HIGH'
            })
    export_endpoints = [e for e in entries if 'export' in e['endpoint'].lower() or 'download' in e['endpoint'].lower()]
    export_users = defaultdict(int)
    for e in export_endpoints:
        export_users[e['user']] += 1
    for user, count in export_users.items():
        if count > 50:
            anomalies.append({
                'type': 'BULK_EXPORT',
                'user': user,
                'count': count,
                'severity': 'CRITICAL'
            })
    hour_distribution = defaultdict(int)
    for e in entries:
        hour = int(e['timestamp'].split('T')[1].split(':')[0])
        hour_distribution[hour] += 1
    off_hours = sum(v for h, v in hour_distribution.items() if h < 6 or h > 22)
    total = sum(hour_distribution.values())
    if total > 0 and off_hours / total > 0.3:
        anomalies.append({
            'type': 'OFF_HOURS_ACTIVITY',
            'off_hours_ratio': f"{off_hours/total:.2%}",
            'severity': 'MEDIUM'
        })
    failed = [e for e in entries if int(e['status']) >= 400]
    if len(failed) / max(len(entries), 1) > 0.1:
        anomalies.append({
            'type': 'HIGH_ERROR_RATE',
            'error_count': len(failed),
            'total': len(entries),
            'severity': 'HIGH'
        })
    return anomalies

if __name__ == '__main__':
    import sys
    if len(sys.argv) < 2:
        print("用法: python lims_api_audit.py <api_log_path>")
        sys.exit(1)
    entries = parse_api_log(sys.argv[1])
    anomalies = detect_api_anomalies(entries)
    report = {
        'total_entries': len(entries),
        'anomalies': anomalies,
        'overall_risk': 'CRITICAL' if any(a['severity'] == 'CRITICAL' for a in anomalies) else 'HIGH' if any(a['severity'] == 'HIGH' for a in anomalies) else 'MEDIUM' if anomalies else 'LOW'
    }
    print(json.dumps(report, indent=2, ensure_ascii=False))

import hashlib
import gzip
import json
import sys
from collections import Counter

VALID_BASES = set('ACGTacgt')

def parse_fasta(fasta_path):
    sequences = []
    current_header = ""
    current_seq = []
    with gzip.open(fasta_path, 'rt') if fasta_path.endswith('.gz') else open(fasta_path, 'r') as f:
        for line in f:
            line = line.strip()
            if line.startswith('>'):
                if current_seq:
                    sequences.append((current_header, ''.join(current_seq)))
                current_header = line[1:]
                current_seq = []
            else:
                current_seq.append(line)
    if current_seq:
        sequences.append((current_header, ''.join(current_seq)))
    return sequences

def verify_base_composition(sequences):
    all_bases = ''.join(seq for _, seq in sequences)
    total = len(all_bases)
    counts = Counter(all_bases.upper())
    invalid = {b: c for b, c in counts.items() if b not in VALID_BASES}
    composition = {b: counts.get(b, 0) / total for b in 'ACGT'}
    at_gc_ratio = (counts.get('A', 0) + counts.get('T', 0)) / max(counts.get('G', 0) + counts.get('C', 0), 1)
    return {
        'total_bases': total,
        'composition': {k: f"{v:.4f}" for k, v in composition.items()},
        'at_gc_ratio': f"{at_gc_ratio:.4f}",
        'invalid_bases': invalid,
        'sequence_count': len(sequences),
        'avg_seq_length': total / max(len(sequences), 1)
    }

def detect_polymer_artifacts(sequences, max_run=10):
    artifacts = []
    for header, seq in sequences:
        for base in 'ACGT':
            run = base * max_run
            pos = seq.upper().find(run)
            if pos != -1:
                artifacts.append({
                    'sequence': header,
                    'polymer': base * max_run,
                    'position': pos,
                    'type': 'POLYMER_RUN'
                })
    return artifacts

def verify_checksum_encoding(sequences, expected_checksum=None):
    all_seq_data = ''.join(seq for _, seq in sequences)
    actual_checksum = hashlib.sha256(all_seq_data.encode()).hexdigest()
    result = {
        'actual_checksum': actual_checksum,
        'total_chars': len(all_seq_data)
    }
    if expected_checksum:
        result['expected_checksum'] = expected_checksum
        result['checksum_match'] = actual_checksum == expected_checksum
        if not actual_checksum == expected_checksum:
            result['ALERT'] = 'DNA存储数据校验和不匹配，数据可能被篡改'
    return result

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("用法: python dna_storage_verify.py <fasta_file> [expected_sha256]")
        sys.exit(1)
    fasta_path = sys.argv[1]
    expected = sys.argv[2] if len(sys.argv) > 2 else None
    seqs = parse_fasta(fasta_path)
    report = {
        'file': fasta_path,
        'base_composition': verify_base_composition(seqs),
        'polymer_artifacts': detect_polymer_artifacts(seqs),
        'checksum': verify_checksum_encoding(seqs, expected)
    }
    print(json.dumps(report, indent=2, ensure_ascii=False))

#!/bin/bash
DB_HOST=$1
DB_NAME="biobank"

echo "=== 生物样本追踪完整性审计 ==="

echo "[1] 检查样本登记时间线异常..."
psql -h "$DB_HOST" -d "$DB_NAME" -c "
SELECT sample_id, created_at, updated_at, 
       EXTRACT(EPOCH FROM (updated_at - created_at)) as age_seconds
FROM samples 
WHERE updated_at > created_at + INTERVAL '1 second'
  AND updated_by != 'SYSTEM'
ORDER BY updated_at DESC
LIMIT 50;
"

echo "[2] 检查批量样本篡改（同一用户短时间修改大量样本）..."
psql -h "$DB_HOST" -d "$DB_NAME" -c "
SELECT updated_by, 
       COUNT(*) as changes,
       MIN(updated_at) as first_change,
       MAX(updated_at) as last_change
FROM samples
WHERE updated_at > NOW() - INTERVAL '24 hours'
GROUP BY updated_by
HAVING COUNT(*) > 100
ORDER BY changes DESC;
"

echo "[3] 检查样本存储温度记录异常..."
psql -h "$DB_HOST" -d "$DB_NAME" -c "
SELECT sensor_id, recorded_at, temperature_celsius,
       CASE 
           WHEN temperature_celsius > 8 THEN 'ALERT_HIGH'
           WHEN temperature_celsius < -196 THEN 'ALERT_CRYO_ERROR'
           WHEN temperature_celsius < -80 AND temperature_celsius > -70 THEN 'WARNING_FREEZER'
           ELSE 'NORMAL'
       END as status
FROM temperature_logs
WHERE recorded_at > NOW() - INTERVAL '7 days'
  AND (temperature_celsius > 8 OR temperature_celsius < -196)
ORDER BY recorded_at DESC;
"

echo "[4] 检查样本分发记录中的异常模式..."
psql -h "$DB_HOST" -d "$DB_NAME" -c "
SELECT d.distributor_name, d.organization,
       COUNT(*) as sample_count,
       COUNT(DISTINCT s.sample_type) as type_diversity,
       MIN(tr.event_time) as first_transfer,
       MAX(tr.event_time) as last_transfer
FROM distributions d
JOIN transfers tr ON d.distribution_id = tr.distribution_id
JOIN samples s ON tr.sample_id = s.sample_id
WHERE tr.event_time > NOW() - INTERVAL '30 days'
GROUP BY d.distributor_name, d.organization
HAVING COUNT(*) > 50
ORDER BY sample_count DESC;
"

echo "[5] 检查样本数据库审计日志..."
if [ -f /var/log/postgresql/biobank_audit.log ]; then
    echo "  最近的DDL操作:"
    grep -i "DDL\|CREATE\|DROP\|ALTER\|TRUNCATE" \
        /var/log/postgresql/biobank_audit.log | tail -20
    echo "  大量DELETE操作:"
    grep "DELETE" /var/log/postgresql/biobank_audit.log | \
        grep -c "DELETE" | awk '{print "[INFO] 总DELETE操作数: " $1}'
fi

import gzip
import json
import sys
import numpy as np
from collections import Counter

def parse_fastq(fastq_path, max_reads=None):
    reads = []
    opener = gzip.open if fastq_path.endswith('.gz') else open
    with opener(fastq_path, 'rt') as f:
        count = 0
        while True:
            header = f.readline().strip()
            if not header:
                break
            seq = f.readline().strip()
            f.readline()
            qual = f.readline().strip()
            reads.append({'header': header, 'seq': seq, 'qual': qual})
            count += 1
            if max_reads and count >= max_reads:
                break
    return reads

def analyze_base_content(reads):
    all_bases = ''.join(r['seq'] for r in reads)
    total = len(all_bases)
    counts = Counter(all_bases)
    composition = {b: counts.get(b, 0) / total for b in 'ACGTN'}
    n_ratio = counts.get('N', 0) / total
    gc = (counts.get('G', 0) + counts.get('C', 0)) / (total - counts.get('N', 0))
    return {
        'total_bases': total,
        'composition': {k: round(v, 6) for k, v in composition.items()},
        'gc_content': round(gc, 4),
        'n_ratio': round(n_ratio, 6),
        'unique_sequences': len(set(r['seq'] for r in reads))
    }

def detect_sequence_bias(reads):
    first_10_bases = Counter()
    last_10_bases = Counter()
    for r in reads:
        if len(r['seq']) >= 10:
            for i, b in enumerate(r['seq'][:10]):
                first_10_bases[(i, b)] += 1
            for i, b in enumerate(r['seq'][-10:]):
                last_10_bases[(i, b)] += 1
    total_reads = len(reads)
    bias_report = {'first_base_bias': {}, 'last_base_bias': {}}
    for i in range(10):
        pos_counts = {b: c for (pos, b), c in first_10_bases.items() if pos == i}
        total_pos = sum(pos_counts.values())
        if total_pos > 0:
            entropy = -sum(
                (c / total_pos) * np.log2(c / total_pos)
                for c in pos_counts.values() if c > 0
            )
            bias_report['first_base_bias'][i] = {
                'entropy': round(entropy, 4),
                'max_freq': round(max(pos_counts.values()) / total_pos, 4) if total_pos else 0
            }
    return bias_report

def detect_homopolymer_run(reads, max_run=20):
    violations = []
    for r in reads[:10000]:
        seq = r['seq']
        for base in 'ACGT':
            run_length = 0
            for b in seq:
                if b == base:
                    run_length += 1
                    if run_length >= max_run:
                        violations.append({
                            'header': r['header'][:50],
                            'polymer': base * run_length,
                            'length': run_length
                        })
                        break
                else:
                    run_length = 0
    return violations

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("用法: python fastq_audit.py <fastq_file> [max_reads]")
        sys.exit(1)
    max_reads = int(sys.argv[2]) if len(sys.argv) > 2 else 500000
    reads = parse_fastq(sys.argv[1], max_reads)
    report = {
        'total_reads': len(reads),
        'base_content': analyze_base_content(reads),
        'sequence_bias': detect_sequence_bias(reads),
        'homopolymer_violations': detect_homopolymer_run(reads),
    }
    n_ratio = report['base_content']['n_ratio']
    if n_ratio > 0.05:
        report['ALERT'] = f'N碱基比例异常偏高: {n_ratio:.2%}，可能数据被污染或篡改'
    gc = report['base_content']['gc_content']
    if gc < 0.3 or gc > 0.7:
        report['ALERT'] = f'GC含量异常: {gc:.2%}，可能物种来源不匹配或数据被替换'
    print(json.dumps(report, indent=2, ensure_ascii=False))

title: 生物信息实验室-LIMS批量数据导出检测
id: biosec-lims-bulk-export-detection
status: experimental
description: 检测LIMS系统中的批量样本数据导出行为，可能指示基因组数据窃取
references:
  - https://www.ncbi.nlm.nih.gov/pmc/articles/PMC9458732/
author: BioSec Forensics Team
date: 2026-07-20
modified: 2026-07-20
tags:
  - attack.exfiltration
  - attack.t1048
  - attack.t1567
  - biosecurity
  - genomics
logsource:
  category: application
  product: lims
  service: api
detection:
  selection_bulk_export:
    endpoint|contains:
      - '/export'
      - '/download'
      - '/batch'
      - '/samples/bulk'
    method:
      - GET
      - POST
  selection_high_volume:
    bytes_transferred|gte: 104857600
  selection_off_hours:
    timestamp|regex: 'T(0[0-5]|2[2-3]):'
  selection_failures:
    status_code|gte: 400
  condition: selection_bulk_export and (selection_high_volume or selection_off_hours or selection_failures)
  timeframe: 1h
  count: 10
level: high
falsepositives:
  - 合法的定期批量数据导出
  - 系统维护期间的数据迁移

title: 生物信息实验室-测序数据文件异常修改
id: biosec-sequencing-data-tampering
status: experimental
description: 检测测序数据文件(BAM/FASTQ/VCF)在非标准操作流程中的修改行为
author: BioSec Forensics Team
date: 2026-07-20
tags:
  - attack.t1565
  - attack.t1005
  - biosecurity
  - genomics
logsource:
  category: file
  product: linux
detection:
  selection_bam_modify:
    TargetFileName|endswith:
      - '.bam'
      - '.bam.bai'
      - '.cram'
    EventID:
      - 11
      - 4663
  selection_fastq_modify:
    TargetFileName|endswith:
      - '.fastq'
      - '.fastq.gz'
      - '.fq'
      - '.fq.gz'
    EventID:
      - 11
      - 4663
  selection_vcf_modify:
    TargetFileName|endswith:
      - '.vcf'
      - '.vcf.gz'
      - '.bcf'
    EventID:
      - 11
      - 4663
  filter_standard_tools:
    Image|contains:
      - 'samtools'
      - 'gatk'
      - 'bwa'
      - 'bowtie'
      - 'nextflow'
      - 'snakemake'
  condition: (selection_bam_modify or selection_fastq_modify or selection_vcf_modify) and not filter_standard_tools
level: medium

#!/bin/bash
LAB_SUBNET="10.50.0.0/16"
LOG_DIR="/var/log/biolab_hunt"
REPORT_FILE="$LOG_DIR/hunt_$(date +%Y%m%d_%H%M%S).json"

mkdir -p "$LOG_DIR"

echo "{"
echo "  \"hunt_timestamp\": \"$(date -Iseconds)\","
echo "  \"hunt_results\": ["

FIRST=true

echo "[HUNT-1] 扫描异常连接到公共基因组数据库..."
PUBLIC_DBS=("ncbi.nlm.nih.gov" "ebi.ac.uk" "uniprot.org" "ensembl.org" "gnomad.broadinstitute.org")
for DB in "${PUBLIC_DBS[@]}"; do
    CONNS=$(ss -tn | grep "$DB" 2>/dev/null | wc -l)
    if [ "$CONNS" -gt 0 ]; then
        if [ "$FIRST" = true ]; then
            FIRST=false
        else
            echo ","
        fi
        echo "    {\"hunt\": \"public_db_access\", \"target\": \"$DB\", \"connections\": $CONNS}"
    fi
done

echo ""
echo "[HUNT-2] 扫描异常进程（可能的恶意生物信息脚本）..."
SUSPICIOUS_PATTERNS=("python.*-c.*import" "perl.*-e.*exec" "curl.*|.*sh" "wget.*|.*bash" "nc.*-e" "python.*socket")
for PATTERN in "${SUSPICIOUS_PATTERNS[@]}"; do
    PIDS=$(pgrep -f "$PATTERN" 2>/dev/null | wc -l)
    if [ "$PIDS" -gt 0 ]; then
        echo ","
        echo "    {\"hunt\": \"suspicious_process\", \"pattern\": \"$PATTERN\", \"count\": $PIDS}"
    fi
done

echo ""
echo "[HUNT-3] 扫描基因组数据文件的异常权限..."
find /data /projects /sequences -name "*.bam" -o -name "*.fastq.gz" -o -name "*.vcf" 2>/dev/null | while read f; do
    PERMS=$(stat -c '%a' "$f" 2>/dev/null)
    if [ "$PERMS" = "777" ] || [ "$PERMS" = "666" ]; then
        echo ","
        echo "    {\"hunt\": \"insecure_file_perms\", \"file\": \"$f\", \"perms\": \"$PERMS\"}"
    fi
done

echo ""
echo "[HUNT-4] 扫描SSH密钥异常..."
for USER_HOME in /home/* /root; do
    AUTH_KEYS="$USER_HOME/.ssh/authorized_keys"
    if [ -f "$AUTH_KEYS" ]; then
        KEY_COUNT=$(grep -c "^ssh-\|^ecdsa-\|^ed25519-" "$AUTH_KEYS" 2>/dev/null)
        if [ "$KEY_COUNT" -gt 3 ]; then
            echo ","
            echo "    {\"hunt\": \"excessive_ssh_keys\", \"user_home\": \"$USER_HOME\", \"key_count\": $KEY_COUNT}"
        fi
    fi
done

echo ""
echo "[HUNT-5] 扫描容器镜像异常..."
if command -v docker &> /dev/null; then
    docker ps --format '{{.Image}} {{.Names}}' 2>/dev/null | while read IMAGE NAME; do
        if echo "$IMAGE" | grep -qE "bioinfo|crispr|pipeline|seq"; then
            DIGEST=$(docker inspect --format='{{index .RepoDigests 0}}' "$IMAGE" 2>/dev/null)
            echo ","
            echo "    {\"hunt\": \"bio_container_running\", \"image\": \"$IMAGE\", \"name\": \"$NAME\", \"digest\": \"$DIGEST\"}"
        fi
    done
fi

echo ""
echo "  ]"
echo "}"

rule Malicious_Bioinfo_Script {
    meta:
        description = "检测包含潜在恶意操作的生物信息分析脚本"
        author = "BioSec Forensics Team"
        date = "2026-07-20"
        reference = "Biosecurity Incident Response"
        severity = "high"
    strings:
        $s1 = "samtools" ascii
        $s2 = "bwa" ascii
        $s3 = "gatk" ascii
        $s4 = "fastq" ascii
        $m1 = "base64" ascii
        $m2 = "eval(" ascii
        $m3 = "exec(" ascii
        $m4 = "__import__" ascii
        $m5 = "subprocess.call" ascii
        $c1 = "socket" ascii
        $c2 = "reverse" ascii
        $c3 = "connect(" ascii
        $c4 = "subprocess.Popen" ascii
        $c5 = "os.system" ascii
    condition:
        filesize < 500KB and
        2 of ($s*) and
        2 of ($m*) and
        1 of ($c*)
}

rule Suspicious_Crispr_Design_Output {
    meta:
        description = "检测可能被篡改的CRISPR设计输出文件"
        author = "BioSec Forensics Team"
        date = "2026-07-20"
        severity = "medium"
    strings:
        $h1 = "sgRNA" ascii nocase
        $h2 = "on_target" ascii nocase
        $h3 = "off_target" ascii nocase
        $h4 = "GC_content" ascii nocase
        $h5 = "chromosome" ascii nocase
        $anomaly1 = "AAAA" ascii
        $anomaly2 = "TTTT" ascii
        $anomaly3 = "GGGGGG" ascii
    condition:
        filesize < 10MB and
        4 of ($h*) and
        2 of ($anomaly*)
}

rule Bioinfo_Data_Exfil_Tool {
    meta:
        description = "检测可能用于基因组数据外泄的工具组合"
        author = "BioSec Forensics Team"
        date = "2026-07-20"
        severity = "critical"
    strings:
        $exfil1 = "rsync" ascii
        $exfil2 = "scp" ascii
        $exfil3 = "s3 cp" ascii
        $exfil4 = "aws s3 sync" ascii
        $exfil5 = "rclone" ascii
        $bio1 = ".fastq" ascii
        $bio2 = ".bam" ascii
        $bio3 = ".vcf" ascii
        $bio4 = ".cr" ascii
        $bio5 = "genom" ascii nocase
    condition:
        filesize < 1MB and
        2 of ($exfil*) and
        2 of ($bio*)
}