1.3.1 更新，新增 --rdp 参数专项分析

bla/core/pipeline.py

@@ -21,6 +21,7 @@
     generate_sarif_report,
 )
 from ..parsers import auto_parse
+from ..parsers.stats import compute_stats
 from ..rules import set_rule_dirs
 from ..utils.helpers import reset_counter, set_syslog_year
 
@@ -51,6 +52,7 @@ class AnalysisOptions:
     rule_dirs: Optional[List[str]] = None
     allowlist_path: Optional[str] = None
     syslog_year: Optional[int] = None
+    rdp_only: bool = False
     outputs: Optional[AnalysisOutputs] = None
 
 
@@ -81,6 +83,7 @@ def parse_files(
     files: List[str],
     jobs: int = 0,
     parser_name: Optional[str] = None,
+    rdp_only: bool = False,
     quiet: bool = False,
     print_fn: Optional[PrintFn] = None,
 ) -> List[ParseResult]:
@@ -95,6 +98,8 @@ def parse_files(
                 emit(f"  [{i}/{len(files)}] 解析: {fname} ...", end=" ", flush=True)
             try:
                 result = auto_parse(fpath, parser_name=parser_name)
+                if rdp_only:
+                    result = _filter_rdp_only_result(result)
                 parse_results.append(result)
                 if not quiet:
                     emit(f"✓ ({result.stats.total} 事件)")
@@ -113,6 +118,8 @@ def parse_files(
                 fname = os.path.basename(fpath)
                 try:
                     result = future.result()
+                    if rdp_only:
+                        result = _filter_rdp_only_result(result)
                     parse_results.append(result)
                     if not quiet:
                         emit(f"  [{done}/{len(files)}] ✓ {fname} ({result.stats.total} 事件)")
@@ -144,6 +151,7 @@ def run_analysis(
         files,
         options.jobs,
         parser_name=options.parser_name,
+        rdp_only=options.rdp_only,
         quiet=quiet,
         print_fn=print_fn,
     )
@@ -204,3 +212,17 @@ def _configure_runtime(options: AnalysisOptions) -> None:
     if options.rule_dirs:
         rule_dirs.extend(options.rule_dirs)
     set_rule_dirs(rule_dirs)
+
+
+def _filter_rdp_only_result(result: ParseResult) -> ParseResult:
+    filtered_events = [event for event in result.events if event.event_id in {"4624", "4625"}]
+    stats = compute_stats(filtered_events)
+    stats.parse_errors = result.stats.parse_errors
+    return ParseResult(
+        file_name=result.file_name,
+        log_type=result.log_type,
+        events=filtered_events,
+        stats=stats,
+        parse_time_ms=result.parse_time_ms,
+        file_size_bytes=result.file_size_bytes,
+    )

bla/output/terminal.py

@@ -91,6 +91,70 @@ def _basename(path: str) -> str:
     return s
 
 
+def _event_sort_key(event: LogEvent) -> tuple:
+    return (
+        event.level.score,
+        event.timestamp or "",
+        event.id,
+    )
+
+
+def _render_logon_event_detail(event: LogEvent, full_evidence: bool) -> List[str]:
+    details = event.details
+    account = details.get("account_name") or event.user or "?"
+    domain = details.get("account_domain") or ""
+    principal = f"{domain}\\{account}" if domain else account
+    source_ip = details.get("source_ip") or event.ip or "-"
+    workstation = details.get("workstation") or "-"
+    logon_type = details.get("LogonType") or "?"
+    logon_label = details.get("logon_type_label") or "未知"
+    process = details.get("logon_process") or details.get("process_name") or "-"
+    auth_package = details.get("auth_package") or "-"
+
+    lines = [
+        (
+            f"        {_level_badge(event.level)} {_fmt_time(event.timestamp)}  "
+            f"账户={principal}  来源IP={source_ip}  登录类型={logon_type}({logon_label})"
+        ),
+        (
+            f"          工作站={workstation}  进程={_truncate_text(process, 28)}  "
+            f"认证={_truncate_text(auth_package, 18)}"
+        ),
+    ]
+
+    if event.event_id == "4625":
+        reason = details.get("failure_reason") or "-"
+        status = details.get("status_code") or "-"
+        sub_status = details.get("sub_status_code") or "-"
+        lines.append(
+            f"          失败原因={_truncate_text(reason, 42)}  Status={status}  SubStatus={sub_status}"
+        )
+    return lines
+
+
+def _render_process_creation_event_detail(event: LogEvent, full_evidence: bool) -> List[str]:
+    details = event.details
+    parent_path = details.get("parent_process") or ""
+    parent = _basename(parent_path) or "(unknown)"
+    child = details.get("child_process") or _basename(details.get("child_path") or event.process or "") or "(unknown)"
+    path = details.get("child_path") or event.process or "-"
+    cmd = details.get("command_line") or details.get("CommandLine") or "-"
+    tags = []
+    for tag in ("malware-indicator", "lolbin", "lsass-dump"):
+        if tag in event.tags:
+            tags.append(tag)
+    tag_text = f"  标记={','.join(tags)}" if tags else ""
+
+    return [
+        (
+            f"        {_level_badge(event.level)} {_fmt_time(event.timestamp)}  "
+            f"父进程={_truncate_text(parent, 20)}  子进程={_truncate_text(child, 20)}{tag_text}"
+        ),
+        f"          路径={_truncate_text(path, 96 if full_evidence else 52)}",
+        f"          命令行={_evidence_text(cmd, full_evidence, 96)}",
+    ]
+
+
 def print_terminal_report(
     parse_results: List[ParseResult],
     summary: AnalysisSummary,
@@ -198,6 +262,18 @@ def print_terminal_report(
                             out.write(f"        SubStatus:   {_fmt_top(event_stats.get('sub_status_codes', []), 'sub_status_code', 5)}\n")
                     out.write("\n")
 
+                logon_recent = sorted(
+                    [event for event in r.events if event.event_id in {"4624", "4625"}],
+                    key=_event_sort_key,
+                    reverse=True,
+                )[:8]
+                if logon_recent:
+                    out.write(f"      {BOLD}具体事件{RESET}  {DIM}(按级别、时间排序，最近 8 条){RESET}\n")
+                    for event in logon_recent:
+                        for line in _render_logon_event_detail(event, full_evidence):
+                            out.write(f"{line}\n")
+                    out.write("\n")
+
             if r.stats.windows_process_creation_stats:
                 pstats = r.stats.windows_process_creation_stats
                 out.write(
@@ -223,6 +299,21 @@ def print_terminal_report(
                     out.write(f"      {idx:>2}. {parent_col:<18} {child_col:<16} {count:>4} {ts_col:<20} {path_col}\n")
                 if pstats.get("suspicious_count", 0) == 0:
                     out.write(f"      {DIM}研判: 进程创建已采集，但未发现明显恶意进程命令行。{RESET}\n")
+                proc_recent = sorted(
+                    [event for event in r.events if event.event_id == "4688"],
+                    key=lambda event: (
+                        1 if any(tag in event.tags for tag in ("malware-indicator", "lolbin", "lsass-dump")) else 0,
+                        event.level.score,
+                        event.timestamp or "",
+                        event.id,
+                    ),
+                    reverse=True,
+                )[:6]
+                if proc_recent:
+                    out.write(f"      {BOLD}具体事件{RESET}  {DIM}(可疑优先，最近 6 条){RESET}\n")
+                    for event in proc_recent:
+                        for line in _render_process_creation_event_detail(event, full_evidence):
+                            out.write(f"{line}\n")
                 out.write("\n")
 
             out.write("\n")

bla/parsers/windows_evtx.py

@@ -647,6 +647,10 @@ def gtag(tag: str) -> str:
     _augment_explicit_credential_details(eid, details)
     _augment_4688_details(eid, details)
 
+    # 4624 若没有可用网络源地址，则默认跳过，不纳入专项统计与输出。
+    if eid == 4624 and not details.get("source_ip"):
+        return None
+
     if eid in (4720, 4722, 4723, 4724, 4725, 4726, 4728, 4729, 4732, 4738, 4756):
         user = details.get("subject_user") or details.get("target_user") or ""
     elif eid == 4776:

bla_cli.py

@@ -491,6 +491,11 @@ def main():
         metavar="YEAR",
         help="指定 Linux syslog/auth.log 这类无年份时间戳使用的年份",
     )
+    parser.add_argument(
+        "--rdp",
+        action="store_true",
+        help="RDP/登录专项模式：仅分析 Windows 安全日志中的 4624/4625 事件",
+    )
     parser.add_argument(
         "--version",
         action="version",
@@ -535,6 +540,7 @@ def main():
                 rule_dirs=args.rules or [],
                 allowlist_path=args.allowlist,
                 syslog_year=args.syslog_year,
+                rdp_only=args.rdp,
             ),
             quiet=False,
             print_fn=print,