文件系统时间戳的错误

#!/usr/bin/env python3
"""
自动化四源 Timestomping 检测引擎
基于 MFT ($SI/$FN)、USN Journal ($J) 和 $LogFile 的交叉验证
依赖: MFTECmd 已预先生成的 CSV 输出文件
"""

import csv
import struct
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass
from typing import Optional, List, Dict, Tuple

# ===== FILETIME 工具函数 =====
WINDOWS_TICK = 10000000
SEC_TO_UNIX_EPOCH = 11644473600

def filetime_to_datetime(ft: int) -> datetime:
"""将 64 位 FILETIME 转换为 datetime 对象"""
if ft == 0:
return datetime(1601, 1, 1)
return datetime.utcfromtimestamp((ft - SEC_TO_UNIX_EPOCH * WINDOWS_TICK) / WINDOWS_TICK)

def check_nanosecond_zero(ft: int) -> bool:
"""检查 FILETIME 的纳秒部分是否全为零 (Timestomping 信号)"""
# FILETIME 的低 32 位包含纳秒精度
return (ft % 10000000) == 0

# ===== 数据结构定义 =====
@dataclass
class MftTimestampRecord:
"""MFT 时间戳记录 — 从 MFTECmd 输出的 MFT CSV 中提取"""
    file_name: str
    full_path: str
    mft_entry: int
    si_created: datetime      # $SI 创建时间
    si_modified: datetime     # $SI 修改时间
    si_accessed: datetime     # $SI 访问时间
    si_mft_changed: datetime  # $SI MFT 变更时间
    fn_created: datetime      # $FN 创建时间
    fn_modified: datetime     # $FN 修改时间
    fn_accessed: datetime     # $FN 访问时间
    fn_mft_changed: datetime  # $FN MFT 变更时间
    si_created_raw: int       # 原始 FILETIME 值 (用于纳秒检测)
    fn_created_raw: int

@dataclass
class UsnRecord:
"""USN Journal 记录"""
    timestamp: datetime
    reason: str
    reason_code: int
    mft_entry: int
    file_name: str

@dataclass
class LogFileRecord:
"""$LogFile 事务记录"""
    timestamp: datetime
    operation: str
    mft_entry: int
    attribute_type: Optional[str]
    lsn: int

@dataclass
class TimestompingAlert:
"""Timestomping 检测告警"""
    file_path: str
    severity: str           # LOW / MEDIUM / HIGH / CRITICAL
    alert_type: str         # 告警类型
    evidence: List[str]     # 证据列表
    si_time: Optional[datetime]
    fn_time: Optional[datetime]
    usn_time: Optional[datetime]
    logfile_time: Optional[datetime]


class TimestompingDetector:
    def __init__(self):
        self.mft_records: Dict[int, MftTimestampRecord] = {}
        self.usn_records: Dict[int, List[UsnRecord]] = defaultdict(list)
        self.logfile_records: Dict[int, List[LogFileRecord]] = defaultdict(list)

    def load_mft_csv(self, csv_path: str):
"""加载 MFTECmd 的 MFT CSV 输出"""
        with open(csv_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
for row in reader:
                entry = int(row.get('EntryNumber', 0))
                si_c = int(row.get('SICreated', 0))
                fn_c = int(row.get('FNCreated', 0))

                record = MftTimestampRecord(
                    file_name=row.get('FileName', ''),
                    full_path=row.get('FullPath', ''),
                    mft_entry=entry,
                    si_created=filetime_to_datetime(si_c),
                    si_modified=filetime_to_datetime(int(row.get('SILastModified', 0))),
                    si_accessed=filetime_to_datetime(int(row.get('SILastAccess', 0))),
                    si_mft_changed=filetime_to_datetime(int(row.get('SIMFTChanged', 0))),
                    fn_created=filetime_to_datetime(fn_c),
                    fn_modified=filetime_to_datetime(int(row.get('FNLastModified', 0))),
                    fn_accessed=filetime_to_datetime(int(row.get('FNLastAccess', 0))),
                    fn_mft_changed=filetime_to_datetime(int(row.get('FNMFTChanged', 0))),
                    si_created_raw=si_c,
                    fn_created_raw=fn_c,
                )
                self.mft_records[entry] = record

    def load_usn_csv(self, csv_path: str):
"""加载 MFTECmd 的 USN Journal CSV 输出"""
        with open(csv_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
for row in reader:
                entry = int(row.get('FileMFTEntryNumber', 0))
                ts = filetime_to_datetime(int(row.get('TimeStamp', 0)))
                reason = row.get('Reason', '')
                reason_code = int(row.get('ReasonCode', 0), 16) if row.get('ReasonCode', '0').startswith('0x') else int(row.get('ReasonCode', 0))

                self.usn_records[entry].append(UsnRecord(
                    timestamp=ts,
                    reason=reason,
                    reason_code=reason_code,
                    mft_entry=entry,
                    file_name=row.get('Name', '')
                ))

    def load_logfile_csv(self, csv_path: str):
"""加载 $LogFile 解析输出"""
        with open(csv_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
for row in reader:
                entry = int(row.get('MftEntry', 0))
                self.logfile_records[entry].append(LogFileRecord(
                    timestamp=datetime.fromisoformat(row.get('Timestamp', '1601-01-01T00:00:00')),
                    operation=row.get('Operation', ''),
                    mft_entry=entry,
                    attribute_type=row.get('AttributeType', None),
                    lsn=int(row.get('LSN', 0))
                ))

    def detect_all(self) -> List[TimestompingAlert]:
"""运行所有检测规则"""
        alerts = []
for entry, mft in self.mft_records.items():
            alerts.extend(self._check_si_fn_discrepancy(mft))
            alerts.extend(self._check_nanosecond_zero(mft))
            alerts.extend(self._check_usn_mft_conflict(mft))
return alerts

    def _check_si_fn_discrepancy(self, mft: MftTimestampRecord) -> List[TimestompingAlert]:
"""检测 $SI 与 $FN 时间戳不一致"""
        alerts = []
        discrepancy_threshold = timedelta(seconds=1)

if abs(mft.si_created - mft.fn_created) > discrepancy_threshold:
            alerts.append(TimestompingAlert(
                file_path=mft.full_path,
                severity="HIGH",
                alert_type="SI_FN_CREATION_MISMATCH",
                evidence=[
                    f"$SI 创建时间: {mft.si_created}",
                    f"$FN 创建时间: {mft.fn_created}",
                    f"时间差: {abs(mft.si_created - mft.fn_created)}"
                ],
                si_time=mft.si_created, fn_time=mft.fn_created,
                usn_time=None, logfile_time=None
            ))
return alerts

    def _check_nanosecond_zero(self, mft: MftTimestampRecord) -> List[TimestompingAlert]:
"""检测时间戳纳秒部分全为零 (系统工具 Timestomping 的信号)"""
        alerts = []
if mft.si_created_raw > 0 and check_nanosecond_zero(mft.si_created_raw):
            alerts.append(TimestompingAlert(
                file_path=mft.full_path,
                severity="MEDIUM",
                alert_type="NANOSECOND_ZERO",
                evidence=[
                    f"$SI 创建时间纳秒全零: {mft.si_created}",
"信号: 可能使用系统工具 (PowerShell/fsutil) 进行 Timestomping"
                ],
                si_time=mft.si_created, fn_time=None,
                usn_time=None, logfile_time=None
            ))
return alerts

    def _check_usn_mft_conflict(self, mft: MftTimestampRecord) -> List[TimestompingAlert]:
"""检测 USN Journal 与 MFT 时间戳冲突"""
        alerts = []
        usn_list = self.usn_records.get(mft.mft_entry, [])

for usn in usn_list:
if usn.reason_code & 0x00000100:  # USN_REASON_FILE_CREATE
if usn.timestamp > mft.si_created:
                    alerts.append(TimestompingAlert(
                        file_path=mft.full_path,
                        severity="CRITICAL",
                        alert_type="USN_CREATE_AFTER_SI_CREATED",
                        evidence=[
                            f"USN FILE_CREATE: {usn.timestamp}",
                            f"$SI 创建时间: {mft.si_created}",
"矛盾: 文件在 MFT 中声称的创建时间早于 USN Journal 记录的实际创建时间!"
                        ],
                        si_time=mft.si_created, fn_time=mft.fn_created,
                        usn_time=usn.timestamp, logfile_time=None
                    ))

if usn.reason_code & 0x00008000:  # USN_REASON_BASIC_INFO_CHANGE
                alerts.append(TimestompingAlert(
                    file_path=mft.full_path,
                    severity="HIGH",
                    alert_type="BASIC_INFO_CHANGE_DETECTED",
                    evidence=[
                        f"USN BASIC_INFO_CHANGE: {usn.timestamp}",
"该记录直接对应 Timestomping 操作",
"USN Journal 中的此时间戳无法通过用户态手段篡改"
                    ],
                    si_time=None, fn_time=None,
                    usn_time=usn.timestamp, logfile_time=None
                ))
return alerts


# ===== 使用示例 =====
if __name__ == "__main__":
    detector = TimestompingDetector()

# 加载由 MFTECmd 预先生成的 CSV 文件
    detector.load_mft_csv("./forensic_output/MFT.csv")
    detector.load_usn_csv("./forensic_output/USNJrnl.csv")

# 检测
    alerts = detector.detect_all()

# 分级输出
print("=" * 70)
print("NTFS Timestomping 检测报告")
print("=" * 70)

    severity_count = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
for alert in sorted(alerts, key=lambda a: ["CRITICAL","HIGH","MEDIUM","LOW"].index(a.severity)):
        severity_count[alert.severity] += 1
print(f"\n[{alert.severity}] {alert.alert_type}")
print(f"    文件: {alert.file_path}")
for evidence in alert.evidence:
print(f"    - {evidence}")

print(f"\n总计: {len(alerts)} 条告警 "
          f"(CRITICAL: {severity_count['CRITICAL']}, "
          f"HIGH: {severity_count['HIGH']}, "
          f"MEDIUM: {severity_count['MEDIUM']})")