From 999515dbbb8bf1adafa8c8b0e66dab3499f85cea Mon Sep 17 00:00:00 2001 From: Henry Hu Date: Tue, 14 Jul 2026 17:49:24 +0700 Subject: [PATCH 1/2] fix: preserve streamlit analysis across stops --- README.md | 10 +- app/api/queue.py | 187 ++++++++++++++++++++- app/database/models.py | 7 + app/database/repository.py | 48 +++++- app/main.py | 135 ++++++++++----- app/ui/background_analysis.py | 263 ++++++++++++++++++++++++++++++ app/ui/cases_tab.py | 220 ++++++++++++++++++++++++- docs/en/USER_MANUAL.md | 16 +- docs/zh-TW/README.md | 5 +- docs/zh-TW/USER_MANUAL.md | 13 +- requirements.txt | 2 +- tests/api/test_queue.py | 89 +++++++++- tests/test_background_analysis.py | 124 ++++++++++++++ tests/test_case_management.py | 3 + tests/test_case_restore.py | 25 +++ tests/test_job_repository.py | 14 ++ tests/test_jobs_schema.py | 10 ++ 17 files changed, 1097 insertions(+), 74 deletions(-) create mode 100644 app/ui/background_analysis.py create mode 100644 tests/test_background_analysis.py diff --git a/README.md b/README.md index ed2c28c..f310a92 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,7 @@ By combining industry-standard network analysis tools (**Zeek**, **Tshark**, **P - **Dedicated MITRE ATT&CK workspace** — evidence-backed technique hypotheses, ATT&CK v19.1 metadata, analyst dispositions, capture coverage, visibility gaps, and Navigator export. - **Capture-quality telemetry** — packet/flow scale, parse ratio, time window, sampling limits, completed stages, and warnings now travel with UI and API results and persist with cases. +- **Durable UI analysis** — Streamlit now submits PCAP work to a process-backed queue, autosaves full evidence to SQLite, and restores recent jobs after a page stop or browser reload. - **Safer PCAP intake** — Streamlit uploads are streamed in bounded chunks, preserve `.pcap`/`.pcapng`, validate magic bytes, enforce batch limits, and remove partial files after rejection. - **Stronger Integrations API** — headless jobs return ATT&CK mappings and capture metrics, IOC feeds carry related technique IDs, readiness checks avoid starting the worker queue, and failed submissions clean up provisional cases and files. - **Evidence without an LLM** — skipping or losing the optional AI narrative no longer hides deterministic packet, IOC, correlation, stage, and warning evidence. @@ -59,9 +60,10 @@ panel walks first-time users through the workflow. ### 2. Progress — transparent 10-stage pipeline -Every stage reports live progress with a skippable per-stage control. PyShark and -Zeek run in parallel, then DNS, TLS, beaconing, and carving fan out concurrently — -you always know what's running and how far it has to go. +Every stage reports durable job progress. PyShark and Zeek run in parallel, then +DNS, TLS, beaconing, and carving fan out concurrently. The work runs outside the +Streamlit page thread, so the upper-right Stop control or a browser reload does +not discard the job; completed evidence is autosaved to Cases and restored. ![Progress tab](docs/images/02-progress.png) @@ -470,7 +472,7 @@ Open `http://localhost:8501` in your browser. 1. **Upload** — Drag and drop one or more `.pcap` files in the Upload tab. Multiple files trigger batch mode with cross-file correlation. 2. **Configure** — Pick an LLM provider (LM Studio / OpenAI / Anthropic), set your home location (Continent > Country > City), OSINT API keys, and optionally a YARA rules directory in the Config tab. 3. **Analyze** — Click **Extract & Analyze** to start the pipeline. -4. **Monitor** — Watch the Progress tab as stages execute: Packet Counting > Parsing + Zeek (parallel) > DNS / TLS / Beaconing / Carving (concurrent) > YARA > OSINT > LLM Report. +4. **Monitor** — Watch the Progress tab as stages execute in a durable background process: Packet Counting > Parsing + Zeek (parallel) > DNS / TLS / Beaconing / Carving (concurrent) > YARA > OSINT > LLM Report. Stopping or reloading the Streamlit page does not discard the job. 5. **Review** — Explore results across Dashboard, MITRE Analysis, LLM Analysis, OSINT, Raw Data, and Cases tabs. 6. **Export** — Download CSV/JSON data, PDF reports, STIX bundles, ATT&CK Navigator layers, or CEF syslog events. diff --git a/app/api/queue.py b/app/api/queue.py index 1ec812e..514f4dd 100644 --- a/app/api/queue.py +++ b/app/api/queue.py @@ -10,6 +10,8 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any +import pandas as pd + from app import config as C from app.database.models import Job, JobStatus from app.database.repository import CaseRepository @@ -27,6 +29,8 @@ WARNING_OSINT_NOT_CONFIGURED = "osint_not_configured" WARNING_OSINT_FAILED = "osint_failed" WARNING_YARA_FAILED = "yara_failed" +WARNING_LLM_NOT_CONFIGURED = "llm_not_configured" +WARNING_LLM_FAILED = "llm_failed" def _load_osint_keys() -> dict[str, str]: @@ -93,6 +97,25 @@ def _sha256_file(path: str) -> str: return h.hexdigest() +def _update_manual_stage(repo: CaseRepository, job_id: str, stage: str, *, completed: bool = False) -> None: + """Publish progress for worker-owned stages that run after ``run_pipeline``.""" + job = repo.get_job(job_id) + if job is None: + return + done = job.progress_done + (1 if completed else 0) + repo.update_job_progress(job_id, stage, min(done, job.progress_total), job.progress_total) + + +def _json_safe_records(value: Any) -> list[dict]: + """Convert a bounded table-like value to JSON-safe records.""" + if isinstance(value, pd.DataFrame): + # ``to_json`` normalizes pandas/numpy scalars, timestamps, and NaN. + return json.loads(value.to_json(orient="records", date_format="iso")) + if isinstance(value, list): + return json.loads(json.dumps(value, default=str)) + return [] + + def _run_yara_stage(result: PipelineResult, opts: dict, job_id: str, repo: CaseRepository) -> dict | None: """Stage 8: YARA over carved files (mirrors app/main.py); returns results, or None when skipped or failed. @@ -100,6 +123,7 @@ def _run_yara_stage(result: PipelineResult, opts: dict, job_id: str, repo: CaseR """ yara_results = None if opts.get("do_yara", True) and result.carved_items: + _update_manual_stage(repo, job_id, "YARA Scanning") try: from app.pipeline.yara_scan import scan_carved_files @@ -115,7 +139,7 @@ def _run_yara_stage(result: PipelineResult, opts: dict, job_id: str, repo: CaseR except Exception: logger.exception("Job %s: yara stage failed", job_id) result.warnings.append(WARNING_YARA_FAILED) - repo.touch_job_heartbeat(job_id) + _update_manual_stage(repo, job_id, "YARA Scanning", completed=True) return yara_results @@ -126,6 +150,7 @@ def _run_osint_stage(result: PipelineResult, opts: dict, job_id: str, repo: Case """ osint_data: dict = {} if opts.get("osint_enabled", True): + _update_manual_stage(repo, job_id, "OSINT enrichment") keys = _load_osint_keys() if not keys: result.warnings.append(WARNING_OSINT_NOT_CONFIGURED) @@ -149,10 +174,108 @@ def _run_osint_stage(result: PipelineResult, opts: dict, job_id: str, repo: Case except Exception: logger.exception("Job %s: osint stage failed", job_id) result.warnings.append(WARNING_OSINT_FAILED) - repo.touch_job_heartbeat(job_id) + _update_manual_stage(repo, job_id, "OSINT enrichment", completed=True) return osint_data +def _load_llm_settings() -> tuple[str, str, str, str, str]: + """Load the active provider settings without putting credentials in a job row.""" + from app.llm import providers as llm_providers + + saved: dict = {} + try: + from app.utils.config_manager import get_config_manager + + saved = get_config_manager().load() or {} + except Exception: + logger.info("ConfigManager unavailable; falling back to LLM env settings") + + provider = saved.get("cfg_llm_provider") or os.getenv("LLM_PROVIDER", C.LLM_PROVIDER_DEFAULT) + if provider not in llm_providers.PROVIDERS: + provider = C.LLM_PROVIDER_DEFAULT + + if provider == llm_providers.PROVIDER_OPENAI: + base_url = saved.get("cfg_openai_base_url") or os.getenv("OPENAI_BASE_URL", "") + api_key = saved.get("cfg_openai_cloud_key") or os.getenv("OPENAI_API_KEY", "") + model = saved.get("cfg_openai_model") or os.getenv("OPENAI_MODEL", C.OPENAI_MODEL_DEFAULT) + elif provider == llm_providers.PROVIDER_ANTHROPIC: + base_url = "" + api_key = saved.get("cfg_anthropic_key") or os.getenv("ANTHROPIC_API_KEY", "") + model = saved.get("cfg_anthropic_model") or os.getenv("ANTHROPIC_MODEL", C.ANTHROPIC_MODEL_DEFAULT) + else: + base_url = saved.get("cfg_llm_endpoint") or os.getenv("LMSTUDIO_BASE_URL", C.LM_BASE_URL) + api_key = saved.get("cfg_openai_key") or os.getenv("LMSTUDIO_API_KEY", C.LM_API_KEY) + model = saved.get("cfg_llm_model") or os.getenv("LMSTUDIO_MODEL", C.LM_MODEL) + + language = saved.get("cfg_llm_language") or os.getenv("LMSTUDIO_LANGUAGE", C.LM_LANGUAGE) + return provider, base_url, api_key, model, language + + +def _run_llm_stage( + result: PipelineResult, + opts: dict, + job_id: str, + repo: CaseRepository, + osint_data: dict, + yara_results: dict | None, +) -> None: + """Stage 10: generate the narrative in the worker so browser stops cannot discard it.""" + # API submissions historically did not run an LLM stage; keep that contract + # unless the caller explicitly opts in (the Streamlit durable-run path does). + if not opts.get("llm_enabled", False): + return + + from app.llm import providers as llm_providers + + _update_manual_stage(repo, job_id, "LLM report") + provider, base_url, api_key, model, language = _load_llm_settings() + if provider in (llm_providers.PROVIDER_OPENAI, llm_providers.PROVIDER_ANTHROPIC) and not api_key: + result.warnings.append(WARNING_LLM_NOT_CONFIGURED) + _update_manual_stage(repo, job_id, "LLM report", completed=True) + return + if not model or (provider == llm_providers.PROVIDER_LMSTUDIO and not base_url): + result.warnings.append(WARNING_LLM_NOT_CONFIGURED) + _update_manual_stage(repo, job_id, "LLM report", completed=True) + return + + context = { + "features": result.features, + "osint": osint_data, + "zeek": {name: _json_safe_records(table) for name, table in result.zeek_tables.items()}, + "beaconing": result.beacon_df_records, + "carved": result.carved_items, + "packet_count": result.packet_count, + "dns_analysis": result.dns_analysis, + "tls_analysis": result.tls_analysis, + "yara_results": yara_results, + "attack_mapping": result.attack_mapping, + "capture_metrics": result.capture_metrics, + "config": { + "limit_packets": opts.get("pyshark_packet_limit"), + "do_pyshark": opts.get("do_pyshark", True), + "do_zeek": opts.get("do_zeek", True), + "do_carve": opts.get("do_carve", True), + "pre_count": opts.get("pre_count", True), + "osint_top_n": opts.get("osint_top_n", C.OSINT_TOP_IPS_DEFAULT), + }, + } + try: + result.summary_narrative = llm_providers.synthesize_report( + provider, + base_url=base_url, + api_key=api_key, + model=model, + context=context, + language=language, + ) + if result.summary_narrative: + result.stages_run.append("llm") + except Exception: + logger.exception("Job %s: LLM stage failed", job_id) + result.warnings.append(WARNING_LLM_FAILED) + _update_manual_stage(repo, job_id, "LLM report", completed=True) + + def _persist_analysis( result: PipelineResult, job: Job, @@ -201,11 +324,26 @@ def _persist_analysis( packet_count=result.packet_count, features=result.features, osint=osint_data or {}, + report=result.summary_narrative or "", yara_results=yara_results, dns_analysis=result.dns_analysis or None, tls_analysis=result.tls_analysis or None, attack_mapping=result.attack_mapping, capture_metrics=result.capture_metrics, + session_artifacts={ + "zeek_tables": {name: _json_safe_records(table) for name, table in (result.zeek_tables or {}).items()}, + "zeek_log_paths": dict(result.zeek_log_paths or {}), + "carved": list(result.carved_items or []), + "beacon_records": list(result.beacon_df_records or []), + "pipeline_warnings": list(result.warnings), + "pipeline_stages": list(result.stages_run), + "duration_seconds": result.duration_seconds, + "rdns_map": { + ip: data["ptr"] + for ip, data in (osint_data or {}).get("ips", {}).items() + if isinstance(data, dict) and data.get("ptr") + }, + }, ) if result.beacon_df_records: analysis.features["beacon_records"] = result.beacon_df_records @@ -216,6 +354,46 @@ def _persist_analysis( result.warnings.append(WARNING_PERSISTENCE_FAILED) +def _run_llm_report_job(job: Job, options_dict: dict, repo: CaseRepository) -> None: + """Regenerate only a persisted analysis report without rerunning packet stages.""" + from app.pipeline.runner import PipelineResult + + analysis_id = options_dict.get("_analysis_id") + analysis = repo.get_analysis(analysis_id) if analysis_id else None + if analysis is None: + raise RuntimeError("The persisted analysis for this report job could not be found.") + + artifacts = analysis.session_artifacts or {} + result = PipelineResult( + case_id=analysis.case_id, + analysis_id=analysis.id, + packet_count=analysis.packet_count, + stages_run=list(artifacts.get("pipeline_stages") or []), + warnings=list(artifacts.get("pipeline_warnings") or []), + dns_analysis=analysis.dns_analysis or {}, + tls_analysis=analysis.tls_analysis or {}, + beacon_df_records=list(artifacts.get("beacon_records") or []), + features=analysis.features or {}, + zeek_tables={ + name: pd.DataFrame.from_records(records) + for name, records in (artifacts.get("zeek_tables") or {}).items() + if isinstance(records, list) + }, + zeek_log_paths=dict(artifacts.get("zeek_log_paths") or {}), + carved_items=list(artifacts.get("carved") or []), + attack_mapping=analysis.attack_mapping, + capture_metrics=analysis.capture_metrics, + ) + _run_llm_stage(result, options_dict, job.id, repo, analysis.osint or {}, analysis.yara_results) + if result.summary_narrative: + analysis.report = result.summary_narrative + artifacts["pipeline_stages"] = list(result.stages_run) + artifacts["pipeline_warnings"] = list(result.warnings) + analysis.session_artifacts = artifacts + repo.save_analysis(analysis) + repo.complete_job(job.id, json.dumps(result.to_dict()).encode("utf-8")) + + def _worker_run(job_id: str, db_path: str, pcap_path: str, options_dict: dict) -> None: """Top-level worker function (must be picklable for ProcessPoolExecutor).""" from app.utils.logger import get_logger @@ -259,6 +437,10 @@ def _on_event(event: ProgressEvent) -> None: options = PipelineOptions(**{k: v for k, v in options_dict.items() if k in PipelineOptions.__dataclass_fields__}) try: + if options_dict.get("_job_type") == "llm_report": + _run_llm_report_job(job, options_dict, repo) + return + result = run_pipeline( pcap_path=pcap_path, case_id=job.case_id, @@ -270,6 +452,7 @@ def _on_event(event: ProgressEvent) -> None: opts = options_dict # raw dict: includes keys PipelineOptions doesn't model (e.g. do_yara) yara_results = _run_yara_stage(result, opts, job_id, repo) osint_data = _run_osint_stage(result, opts, job_id, repo) + _run_llm_stage(result, opts, job_id, repo, osint_data, yara_results) _persist_analysis(result, job, pcap_path, osint_data, yara_results, repo) diff --git a/app/database/models.py b/app/database/models.py index 19f0c70..944dde5 100644 --- a/app/database/models.py +++ b/app/database/models.py @@ -158,6 +158,11 @@ class Analysis: tls_analysis: dict | None = None attack_mapping: dict | None = None capture_metrics: dict | None = None + # UI-only evidence that is expensive to regenerate (bounded Zeek tables, + # carved-file metadata, beacon rows, stage warnings, and run log paths). + # Keeping it separate from ``features`` avoids leaking persistence details + # into the analysis/scoring contracts. + session_artifacts: dict | None = None iocs: list[IOC] = field(default_factory=list) def to_dict(self) -> dict: @@ -176,6 +181,7 @@ def to_dict(self) -> dict: "tls_analysis": self.tls_analysis, "attack_mapping": self.attack_mapping, "capture_metrics": self.capture_metrics, + "session_artifacts": self.session_artifacts, "iocs": [ioc.to_dict() for ioc in self.iocs], } @@ -202,6 +208,7 @@ def from_dict(cls, data: dict) -> "Analysis": tls_analysis=data.get("tls_analysis"), attack_mapping=data.get("attack_mapping"), capture_metrics=data.get("capture_metrics"), + session_artifacts=data.get("session_artifacts"), iocs=iocs, ) diff --git a/app/database/repository.py b/app/database/repository.py index 1f4694f..a5268ff 100644 --- a/app/database/repository.py +++ b/app/database/repository.py @@ -73,7 +73,8 @@ def _init_schema(self): dns_json TEXT, tls_json TEXT, attack_mapping_json TEXT, - capture_metrics_json TEXT + capture_metrics_json TEXT, + session_artifacts_json TEXT ); -- IOCs extracted from analyses @@ -140,7 +141,7 @@ def _init_schema(self): # Existing case databases predate ATT&CK and capture-quality # persistence. Add the columns in place so upgrades do not erase # prior investigations. - for column in ("attack_mapping_json", "capture_metrics_json"): + for column in ("attack_mapping_json", "capture_metrics_json", "session_artifacts_json"): try: conn.execute(f"ALTER TABLE analyses ADD COLUMN {column} TEXT") # noqa: S608 — fixed column names except sqlite3.OperationalError as exc: @@ -419,6 +420,9 @@ def save_analysis(self, analysis: Analysis) -> str: tls_json = self._compress_json(analysis.tls_analysis) if analysis.tls_analysis else None attack_mapping_json = self._compress_json(analysis.attack_mapping) if analysis.attack_mapping else None capture_metrics_json = self._compress_json(analysis.capture_metrics) if analysis.capture_metrics else None + session_artifacts_json = ( + self._compress_json(analysis.session_artifacts) if analysis.session_artifacts else None + ) params = ( analysis.id, @@ -435,14 +439,15 @@ def save_analysis(self, analysis: Analysis) -> str: tls_json, attack_mapping_json, capture_metrics_json, + session_artifacts_json, ) conn.execute( """ INSERT INTO analyses (id, case_id, pcap_path, pcap_hash, packet_count, analyzed_at, features_json, osint_json, report_md, yara_json, dns_json, tls_json, - attack_mapping_json, capture_metrics_json) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + attack_mapping_json, capture_metrics_json, session_artifacts_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET case_id = excluded.case_id, pcap_path = excluded.pcap_path, @@ -456,7 +461,8 @@ def save_analysis(self, analysis: Analysis) -> str: dns_json = excluded.dns_json, tls_json = excluded.tls_json, attack_mapping_json = excluded.attack_mapping_json, - capture_metrics_json = excluded.capture_metrics_json + capture_metrics_json = excluded.capture_metrics_json, + session_artifacts_json = excluded.session_artifacts_json """, params, ) @@ -736,6 +742,7 @@ def _row_to_analysis(self, row: dict, conn: sqlite3.Connection) -> Analysis: tls_analysis = self._decompress_json(row.get("tls_json")) attack_mapping = self._decompress_json(row.get("attack_mapping_json")) capture_metrics = self._decompress_json(row.get("capture_metrics_json")) + session_artifacts = self._decompress_json(row.get("session_artifacts_json")) # Load IOCs ioc_rows = conn.execute("SELECT * FROM iocs WHERE analysis_id = ?", (row["id"],)).fetchall() @@ -765,6 +772,7 @@ def _row_to_analysis(self, row: dict, conn: sqlite3.Connection) -> Analysis: tls_analysis=tls_analysis, attack_mapping=attack_mapping, capture_metrics=capture_metrics, + session_artifacts=session_artifacts, iocs=iocs, ) @@ -871,6 +879,36 @@ def get_job(self, job_id: str) -> Job | None: finally: conn.close() + def list_jobs( + self, + *, + case_id: str | None = None, + statuses: list[JobStatus] | None = None, + limit: int = 100, + ) -> list[Job]: + """List recent jobs, optionally scoped to a case and lifecycle states.""" + clauses: list[str] = [] + params: list[Any] = [] + if case_id: + clauses.append("case_id = ?") + params.append(case_id) + if statuses: + placeholders = ", ".join("?" for _ in statuses) + clauses.append(f"status IN ({placeholders})") # noqa: S608 - placeholders only + params.extend(status.value for status in statuses) + + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + params.append(max(1, min(int(limit), 1000))) + conn = self._get_conn() + try: + rows = conn.execute( + f"SELECT * FROM jobs {where} ORDER BY submitted_at DESC LIMIT ?", # noqa: S608 - fixed clauses + params, + ).fetchall() + return [Job.from_dict(dict(row)) for row in rows] + finally: + conn.close() + def update_job_status( self, job_id: str, diff --git a/app/main.py b/app/main.py index 6a33280..c67ecc0 100644 --- a/app/main.py +++ b/app/main.py @@ -25,7 +25,13 @@ PhaseTracker, end_run, is_run_active, - reset_run_state, +) +from app.ui.background_analysis import ( + BACKGROUND_RUN_KEY, + find_recoverable_background_run, + render_background_progress, + submit_background_analysis, + submit_background_report, ) from app.ui.charts import ( build_sankey_html, @@ -39,7 +45,7 @@ plot_traffic_timeline_heatmap, plot_world_map, ) -from app.ui.config_ui import init_config_defaults, render_config_tab +from app.ui.config_ui import init_config_defaults, render_config_tab, save_config from app.ui.layout import ( analysis_has_run, inject_css, @@ -270,16 +276,29 @@ def _run_single_pcap_pipeline( # --- RE-RUN TRIGGER LOGIC --- if st.session_state.get("trigger_llm_rerun"): - # Clear and reset LLM phase - st.session_state["run_active"] = True - llm_slug = make_slug("LLM report") - st.session_state[f"done_{llm_slug}"] = False - st.session_state[f"skip_{llm_slug}"] = False - st.session_state["report"] = None - st.session_state["llm_status"] = None - # Consume the trigger - st.session_state["trigger_llm_rerun"] = False - st.rerun() + restored_ids = st.session_state.get("restored_analysis_ids") or [] + if restored_ids: + save_config() + try: + st.session_state[BACKGROUND_RUN_KEY] = submit_background_report(restored_ids) + except Exception as exc: + st.error(f"Could not queue report regeneration: {exc}") + else: + st.session_state["report"] = None + st.session_state["llm_status"] = None + st.session_state["trigger_llm_rerun"] = False + st.rerun() + else: + # Legacy in-memory runs have no persisted analysis to hand to the + # report-only worker, so retain the original fallback behavior. + st.session_state["run_active"] = True + llm_slug = make_slug("LLM report") + st.session_state[f"done_{llm_slug}"] = False + st.session_state[f"skip_{llm_slug}"] = False + st.session_state["report"] = None + st.session_state["llm_status"] = None + st.session_state["trigger_llm_rerun"] = False + st.rerun() init_config_defaults() # ---------------------- Dependency pre-flight check ---------------------- @@ -358,6 +377,13 @@ def _run_single_pcap_pipeline( if k not in st.session_state: st.session_state[k] = v +# Browser reloads create a new Streamlit session, but durable job rows remain. +# Reattach the latest recent UI-owned run when there is no live analysis yet. +if BACKGROUND_RUN_KEY not in st.session_state and not analysis_has_run(): + recovered_run = find_recoverable_background_run() + if recovered_run: + st.session_state[BACKGROUND_RUN_KEY] = recovered_run + # ---------------------- 1) Upload ---------------------- with tab_upload: st.subheader("1) Load PCAP") @@ -468,45 +494,72 @@ def _run_single_pcap_pipeline( ("LLM report", True), ] + run_llm = st.checkbox( + "Generate LLM report in the background", + value=bool(st.session_state.get("cfg_run_llm", True)), + key="cfg_run_llm", + help="Disable this when you only need deterministic packet, Zeek, YARA, and OSINT results.", + ) start = st.button("Extract & Analyze", type="primary", width="stretch") if start: if not pcap_path or not pathlib.Path(pcap_path).exists(): st.error("Please upload a PCAP or provide a valid path.") st.stop() - reset_run_state([t for (t, enabled) in phases if enabled]) - st.session_state.update( - { - "features": None, - "osint": None, - "report": None, - "llm_status": None, - "beacon_df": pd.DataFrame(), - "zeek_tables": {}, - "carved": [], - "__total_pkts": None, - "__pcap_path": pcap_path, - "__pcap_paths": pcap_paths or [pcap_path], - "dns_analysis": None, - "tls_analysis": None, - "attack_mapping": None, - "capture_metrics": None, - "pipeline_warnings": [], - "pipeline_stages": [], - "yara_results": None, - "correlations": None, - "flow_asymmetry": None, - "port_anomalies": None, - "__batch_result": None, - } - ) - st.toast("Analysis started — follow progress in the Progress tab", icon="🚀") - st.success("Analysis started. Switch to the **Progress** tab to monitor.") + try: + limit_packets = int(st.session_state.get("cfg_limit_packets", C.DEFAULT_PYSHARK_LIMIT)) or None + except (ValueError, TypeError): + limit_packets = C.DEFAULT_PYSHARK_LIMIT + try: + osint_top_n = int(st.session_state.get("cfg_osint_top_ips", C.OSINT_TOP_IPS_DEFAULT) or 0) + except (ValueError, TypeError): + osint_top_n = C.OSINT_TOP_IPS_DEFAULT + + # Persist encrypted provider settings before the worker process loads + # them. Job rows contain only non-sensitive execution options. + if not save_config(): + st.warning( + "Settings could not be saved; the background job will use environment/default provider settings." + ) + try: + background_run = submit_background_analysis( + pcap_paths or [pcap_path], + { + "osint_enabled": True, + "llm_enabled": run_llm, + "do_pyshark": do_pyshark, + "do_zeek": do_zeek, + "do_carve": do_carve, + "do_yara": do_yara, + "pre_count": pre_count, + "pyshark_packet_limit": limit_packets, + "osint_top_n": osint_top_n, + }, + ) + except Exception as exc: + st.error(f"Could not start the background analysis: {exc}") + st.stop() + end_run() + st.session_state[BACKGROUND_RUN_KEY] = background_run + st.session_state["__pcap_path"] = pcap_path + st.session_state["__pcap_paths"] = pcap_paths or [pcap_path] + st.session_state["__batch_mode"] = len(st.session_state["__pcap_paths"]) > 1 + st.toast("Analysis safely queued in the background", icon="🚀") + st.success("Analysis started. It will continue even if this Streamlit page is stopped or reloaded.") st.rerun() # ---------------------- 2) Progress ---------------------- with tab_progress: progress_panel = make_progress_panel(st.container()) - if is_run_active(): + background_run = st.session_state.get(BACKGROUND_RUN_KEY) + if background_run: + + @st.fragment(run_every="2s") + def _background_progress_fragment(): + if render_background_progress(st.session_state[BACKGROUND_RUN_KEY]): + st.rerun() + + _background_progress_fragment() + elif is_run_active(): pcap_path = st.session_state.get("__pcap_path") pcap_paths = st.session_state.get("__pcap_paths") or ([pcap_path] if pcap_path else []) batch_mode = st.session_state.get("__batch_mode", False) and len(pcap_paths) > 1 diff --git a/app/ui/background_analysis.py b/app/ui/background_analysis.py new file mode 100644 index 0000000..66ffad4 --- /dev/null +++ b/app/ui/background_analysis.py @@ -0,0 +1,263 @@ +"""Durable Streamlit analysis jobs. + +Long-running PCAP work must not execute inside Streamlit's script thread: the +browser's Stop control intentionally terminates that thread. This module sends +the work to the existing process-backed queue, monitors SQLite state, and +restores persisted results into the UI when every job finishes. +""" + +from __future__ import annotations + +import json +import logging +import os +import pathlib +import uuid +from datetime import datetime, timedelta +from functools import lru_cache +from typing import Any + +import streamlit as st + +from app.api.queue import InProcessJobQueue, JobSubmission +from app.database.models import Case, CaseStatus, Job, JobStatus, Severity +from app.database.repository import CaseRepository +from app.ui.cases_tab import restore_analyses_to_session + +logger = logging.getLogger(__name__) + +BACKGROUND_RUN_KEY = "background_analysis_run" +BACKGROUND_ORIGIN = "streamlit" + + +@lru_cache(maxsize=1) +def get_background_repo() -> CaseRepository: + """Return the process-wide repository used by Streamlit job monitoring.""" + return CaseRepository() + + +@lru_cache(maxsize=1) +def get_background_queue() -> InProcessJobQueue: + """Create one worker pool for the Streamlit server, not one per rerun/session.""" + try: + workers = max(1, int(os.getenv("PCAP_HUNTER_UI_WORKERS", "2"))) + depth = max(1, int(os.getenv("PCAP_HUNTER_UI_QUEUE_DEPTH", "100"))) + except ValueError: + workers, depth = 2, 100 + return InProcessJobQueue(get_background_repo(), max_workers=workers, queue_depth=depth) + + +def _is_streamlit_job(job: Job) -> bool: + return _job_options(job).get("_origin") == BACKGROUND_ORIGIN + + +def _job_options(job: Job) -> dict[str, Any]: + try: + options = json.loads(job.options_json or "{}") + except (TypeError, json.JSONDecodeError): + return {} + return options if isinstance(options, dict) else {} + + +def submit_background_analysis(pcap_paths: list[str], options: dict[str, Any]) -> dict[str, Any]: + """Create an autosaved case and enqueue one durable job per PCAP.""" + paths = [str(pathlib.Path(path)) for path in pcap_paths if path] + if not paths: + raise ValueError("At least one PCAP path is required.") + + repo = get_background_repo() + queue = get_background_queue() + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M") + if len(paths) == 1: + title = f"{pathlib.Path(paths[0]).name} — {timestamp}" + else: + title = f"Batch analysis ({len(paths)} PCAPs) — {timestamp}" + case = Case( + title=title, + description=( + "Automatically saved background analysis. The job continues if the Streamlit page is stopped or reloaded." + ), + status=CaseStatus.IN_PROGRESS, + severity=Severity.MEDIUM, + tags=["autosaved", "background"], + ) + case_id = repo.create_case(case) + + job_ids: list[str] = [] + durable_options = dict(options) + run_id = uuid.uuid4().hex + durable_options.update({"_origin": BACKGROUND_ORIGIN, "_run_id": run_id, "_batch_size": len(paths)}) + try: + for index, path in enumerate(paths): + per_file_options = dict(durable_options) + per_file_options["_batch_index"] = index + job_ids.append(queue.enqueue(JobSubmission(case_id=case_id, pcap_path=path, options=per_file_options))) + except Exception: + logger.exception("Could not enqueue the complete Streamlit background run") + if not job_ids: + repo.delete_case(case_id) + raise + + return { + "case_id": case_id, + "job_ids": job_ids, + "pcap_paths": paths, + "loaded": False, + "submitted_at": datetime.now().isoformat(), + "run_id": run_id, + } + + +def submit_background_report(analysis_ids: list[str]) -> dict[str, Any]: + """Queue report-only regeneration for one or more persisted analyses.""" + repo = get_background_repo() + queue = get_background_queue() + analyses = [repo.get_analysis(analysis_id) for analysis_id in analysis_ids] + if not analyses or any(analysis is None for analysis in analyses): + raise ValueError("The saved analysis needed for report regeneration could not be found.") + analyses = [analysis for analysis in analyses if analysis is not None] + case_id = analyses[0].case_id + if any(analysis.case_id != case_id for analysis in analyses): + raise ValueError("A report batch must belong to one case.") + + run_id = uuid.uuid4().hex + job_ids = [] + for index, analysis in enumerate(analyses): + options = { + "_origin": BACKGROUND_ORIGIN, + "_run_id": run_id, + "_job_type": "llm_report", + "_analysis_id": analysis.id, + "_batch_size": len(analyses), + "_batch_index": index, + "llm_enabled": True, + } + job_ids.append( + queue.enqueue(JobSubmission(case_id=case_id, pcap_path=analysis.pcap_path or analysis.id, options=options)) + ) + case = repo.get_case(case_id) + if case is not None: + case.status = CaseStatus.IN_PROGRESS + repo.update_case(case) + return { + "case_id": case_id, + "job_ids": job_ids, + "pcap_paths": [analysis.pcap_path for analysis in analyses], + "loaded": False, + "submitted_at": datetime.now().isoformat(), + "run_id": run_id, + "report_only": True, + } + + +def find_recoverable_background_run(max_age_days: int = 7) -> dict[str, Any] | None: + """Find the latest UI-owned run after a browser reload resets Session State.""" + repo = get_background_repo() + jobs = [job for job in repo.list_jobs(limit=500) if _is_streamlit_job(job)] + if not jobs: + return None + + cutoff = datetime.now() - timedelta(days=max_age_days) + recent = [job for job in jobs if (job.submitted_at or datetime.min) >= cutoff] + if not recent: + return None + + active = [job for job in recent if job.status in (JobStatus.QUEUED, JobStatus.RUNNING)] + anchor = max(active or recent, key=lambda job: job.submitted_at or datetime.min) + anchor_options = _job_options(anchor) + run_id = anchor_options.get("_run_id") + case_jobs = sorted( + [ + job + for job in recent + if job.case_id == anchor.case_id and (not run_id or _job_options(job).get("_run_id") == run_id) + ], + key=lambda job: job.submitted_at or datetime.min, + ) + return { + "case_id": anchor.case_id, + "job_ids": [job.id for job in case_jobs], + "pcap_paths": [job.pcap_path for job in case_jobs], + "loaded": False, + "submitted_at": (anchor.submitted_at or datetime.now()).isoformat(), + "recovered": True, + "run_id": run_id, + "report_only": anchor_options.get("_job_type") == "llm_report", + } + + +def _load_completed_analyses(repo: CaseRepository, jobs: list[Job]): + analyses = [] + for job in jobs: + try: + result = json.loads(job.result_json or "{}") + except (TypeError, json.JSONDecodeError) as exc: + raise RuntimeError(f"Job {job.id} completed without a readable result.") from exc + analysis_id = result.get("analysis_id") + if not analysis_id: + raise RuntimeError(f"Job {job.id} completed but its analysis was not persisted.") + analysis = repo.get_analysis(analysis_id) + if analysis is None: + raise RuntimeError(f"Persisted analysis {analysis_id} could not be found.") + analyses.append(analysis) + return analyses + + +def render_background_progress(run: dict[str, Any]) -> bool: + """Render current job state and restore results; return True after first restore.""" + repo = get_background_repo() + jobs = [repo.get_job(job_id) for job_id in run.get("job_ids", [])] + if not jobs or any(job is None for job in jobs): + st.error("The saved background job record could not be found. Check Cases for any completed evidence.") + return False + jobs = [job for job in jobs if job is not None] + + total_units = sum(max(job.progress_total, 1) for job in jobs) + completed_units = sum(min(job.progress_done, max(job.progress_total, 1)) for job in jobs) + percent = int(completed_units / total_units * 100) if total_units else 0 + st.progress(min(percent, 100), text=f"Background analysis: {percent}%") + st.caption( + "This analysis runs outside the Streamlit page. The upper-right Stop control only pauses this display; " + "job progress remains in the case database and the final evidence is autosaved when processing completes." + ) + + for index, job in enumerate(jobs, start=1): + filename = pathlib.Path(job.pcap_path).name + stage = job.progress_stage or job.status.value.replace("_", " ").title() + st.write(f"**{index}/{len(jobs)} — {filename}:** {stage} ({job.status.value})") + + failed = [job for job in jobs if job.status in (JobStatus.FAILED, JobStatus.CANCELLED)] + if failed: + for job in failed: + detail = job.error_detail or job.error_code or job.status.value + st.error(f"{pathlib.Path(job.pcap_path).name}: {detail}") + st.info("Any analyses that completed before the failure remain available in Cases.") + return False + + if not all(job.status == JobStatus.DONE for job in jobs): + running = sum(job.status == JobStatus.RUNNING for job in jobs) + queued = sum(job.status == JobStatus.QUEUED for job in jobs) + st.info(f"Analysis is continuing safely in the background ({running} running, {queued} queued).") + return False + + if run.get("loaded"): + st.success("Analysis complete and restored. Review Dashboard, MITRE Analysis, LLM Analysis, and Raw Data.") + return False + + try: + analyses = _load_completed_analyses(repo, jobs) + restore_analyses_to_session(analyses) + except Exception as exc: + logger.exception("Could not restore completed background analysis") + st.error(f"The job completed, but the workbench could not restore its results: {exc}") + st.info("The persisted evidence is still available in Cases.") + return False + + case = repo.get_case(run["case_id"]) + if case is not None and case.status == CaseStatus.IN_PROGRESS: + case.status = CaseStatus.OPEN + repo.update_case(case) + run["loaded"] = True + st.session_state[BACKGROUND_RUN_KEY] = run + st.success("Analysis complete. Results were autosaved and restored into the workbench.") + return True diff --git a/app/ui/cases_tab.py b/app/ui/cases_tab.py index 60eedc1..59e7883 100644 --- a/app/ui/cases_tab.py +++ b/app/ui/cases_tab.py @@ -2,14 +2,19 @@ from __future__ import annotations +import json from datetime import datetime import pandas as pd import streamlit as st from app.analysis.flow_aggregates import compute_flow_aggregates +from app.analysis.visibility import build_capture_metrics from app.database import Analysis, Case, CaseRepository, CaseStatus, IOCType, Severity +from app.pipeline.batch import BatchProcessor, PCAPResult from app.ui.colors import severity_color +from app.ui.mitre_page import build_attack_mapping +from app.utils.common import uniq_sorted from app.utils.logger import get_logger logger = get_logger(__name__) @@ -73,32 +78,229 @@ def _restore_analysis_to_session(analysis: Analysis) -> None: # the new column are handled by the dedicated page's lazy recomputation. st.session_state["attack_mapping"] = analysis.attack_mapping st.session_state["capture_metrics"] = analysis.capture_metrics - st.session_state["pipeline_warnings"] = [] - st.session_state["pipeline_stages"] = [] + session_artifacts = analysis.session_artifacts or {} + st.session_state["pipeline_warnings"] = list(session_artifacts.get("pipeline_warnings") or []) + st.session_state["pipeline_stages"] = list(session_artifacts.get("pipeline_stages") or []) st.session_state["yara_results"] = analysis.yara_results # Model default for report is "" but the app's no-report sentinel is None. st.session_state["report"] = analysis.report or None st.session_state["llm_status"] = "generated" if analysis.report else None - # Everything below isn't persisted on Analysis — reset it all, otherwise the - # dashboard mixes this case's data with leftovers from the previous live run. - st.session_state["beacon_df"] = pd.DataFrame() + # New durable background analyses persist the bounded UI evidence below. + # Legacy cases have no session_artifacts column value and still take the + # safe empty-state path instead of mixing in data from a previous capture. + beacon_records = session_artifacts.get("beacon_records") or features.get("beacon_records") or [] + st.session_state["beacon_df"] = pd.DataFrame.from_records(beacon_records) st.session_state["ja3_df"] = pd.DataFrame() st.session_state["ja3_analysis"] = {} - st.session_state["zeek_tables"] = {} - st.session_state["carved"] = [] + st.session_state["zeek_tables"] = { + name: pd.DataFrame.from_records(records) + for name, records in (session_artifacts.get("zeek_tables") or {}).items() + if isinstance(records, list) + } + st.session_state["carved"] = list(session_artifacts.get("carved") or []) # None (not []) so empty-state rendering says "not available — re-run", rather # than a false "ran clean" for results that simply weren't persisted. st.session_state["correlations"] = None st.session_state["flow_asymmetry"] = None st.session_state["port_anomalies"] = None - st.session_state["rdns_map"] = {} + st.session_state["rdns_map"] = dict(session_artifacts.get("rdns_map") or {}) + st.session_state["__pcap_path"] = analysis.pcap_path + st.session_state["__pcap_paths"] = [analysis.pcap_path] if analysis.pcap_path else [] + st.session_state["__batch_mode"] = False + st.session_state["__batch_result"] = None st.session_state["filter_ips"] = set() st.session_state["filter_protos"] = set() st.session_state["filter_time"] = None st.session_state["restored_analysis_id"] = analysis.id + st.session_state["restored_analysis_ids"] = [analysis.id] + if session_artifacts: + _restore_expensive_derived_state([analysis]) logger.info("Restored analysis %s into session state", analysis.id) +def _analysis_to_pcap_result(analysis: Analysis) -> PCAPResult: + """Convert a persisted analysis back to the production-shape batch result.""" + artifacts = analysis.session_artifacts or {} + zeek_tables = { + name: pd.DataFrame.from_records(records) + for name, records in (artifacts.get("zeek_tables") or {}).items() + if isinstance(records, list) + } + beacon_records = artifacts.get("beacon_records") or (analysis.features or {}).get("beacon_records") or [] + return PCAPResult( + path=analysis.pcap_path, + filename=analysis.pcap_path.rsplit("/", 1)[-1] or analysis.id, + features=analysis.features or {}, + zeek_tables=zeek_tables, + zeek_log_paths=dict(artifacts.get("zeek_log_paths") or {}), + rdns_map=dict(artifacts.get("rdns_map") or {}), + carved_items=list(artifacts.get("carved") or []), + osint=analysis.osint or {}, + beacon_df=pd.DataFrame.from_records(beacon_records), + dns_analysis=analysis.dns_analysis or {}, + tls_analysis=analysis.tls_analysis or {}, + packet_count=analysis.packet_count, + duration_seconds=float(artifacts.get("duration_seconds") or 0), + stages_run=list(artifacts.get("pipeline_stages") or []), + warnings=list(artifacts.get("pipeline_warnings") or []), + ) + + +def _merge_yara_results(analyses: list[Analysis]) -> dict | None: + per_file = [] + matches = [] + for analysis in analyses: + if not analysis.yara_results: + continue + per_file.append({"pcap_path": analysis.pcap_path, "result": analysis.yara_results}) + if isinstance(analysis.yara_results, dict): + matches.extend(analysis.yara_results.get("matches") or []) + if not per_file: + return None + return {"matches": matches, "per_file": per_file} + + +def _current_session_artifacts() -> dict: + """Capture the bounded evidence needed to reopen the current UI result.""" + zeek_tables = {} + for name, table in (st.session_state.get("zeek_tables") or {}).items(): + if isinstance(table, pd.DataFrame): + zeek_tables[name] = json.loads(table.to_json(orient="records", date_format="iso")) + beacon_df = st.session_state.get("beacon_df") + beacon_records = ( + json.loads(beacon_df.to_json(orient="records", date_format="iso")) + if isinstance(beacon_df, pd.DataFrame) + else [] + ) + return { + "zeek_tables": zeek_tables, + "zeek_log_paths": dict(st.session_state.get("zeek_log_paths") or {}), + "carved": list(st.session_state.get("carved") or []), + "beacon_records": beacon_records, + "pipeline_warnings": list(st.session_state.get("pipeline_warnings") or []), + "pipeline_stages": list(st.session_state.get("pipeline_stages") or []), + "rdns_map": dict(st.session_state.get("rdns_map") or {}), + } + + +def _restore_expensive_derived_state(analyses: list[Analysis]) -> None: + """Rebuild inexpensive cross-links after durable evidence is restored.""" + features = st.session_state.get("features") or {} + beacon_df = st.session_state.get("beacon_df") + try: + from app.analysis.correlation import correlate_indicators + from app.analysis.flow_analysis import detect_flow_asymmetry, detect_port_anomalies + + st.session_state["correlations"] = correlate_indicators( + features=features, + osint=st.session_state.get("osint") or {}, + beacon_df=beacon_df if isinstance(beacon_df, pd.DataFrame) else pd.DataFrame(), + dns_analysis=st.session_state.get("dns_analysis"), + tls_analysis=st.session_state.get("tls_analysis"), + yara_results=st.session_state.get("yara_results"), + ) + flows = features.get("flows") or [] + if flows: + st.session_state["flow_asymmetry"] = detect_flow_asymmetry(flows) + st.session_state["port_anomalies"] = detect_port_anomalies(flows) + except Exception as exc: + logger.warning("Could not rebuild restored correlations: %s", exc) + + log_paths = [ + dict((analysis.session_artifacts or {}).get("zeek_log_paths") or {}) + for analysis in analyses + if (analysis.session_artifacts or {}).get("zeek_log_paths") + ] + try: + if len(log_paths) > 1: + from app.pipeline.ja3 import extract_ja3_from_multiple_runs + + ja3_df, ja3_analysis = extract_ja3_from_multiple_runs(log_paths) + elif log_paths: + from app.pipeline.zeek import extract_ja3_from_zeek_tables + + ja3_df, ja3_analysis = extract_ja3_from_zeek_tables(log_paths[0]) + else: + ja3_df, ja3_analysis = pd.DataFrame(), {} + st.session_state["ja3_df"] = ja3_df + st.session_state["ja3_analysis"] = ja3_analysis + except Exception as exc: + logger.warning("Could not rebuild restored JA3 state: %s", exc) + + +def restore_analyses_to_session(analyses: list[Analysis]) -> None: + """Restore one or more completed background analyses into the workbench.""" + analyses = [analysis for analysis in analyses if analysis is not None] + if not analyses: + raise ValueError("No persisted analyses were available to restore.") + if len(analyses) == 1: + _restore_analysis_to_session(analyses[0]) + return + + processor = BatchProcessor([]) + for analysis in analyses: + processor.add_result(_analysis_to_pcap_result(analysis)) + batch_result = processor.merge_all() + + artifact_values: dict[str, set] = {} + all_flows: list[dict] = [] + for result in batch_result.pcap_results: + all_flows.extend((result.features or {}).get("flows") or []) + for key, values in ((result.features or {}).get("artifacts") or {}).items(): + if isinstance(values, list): + artifact_values.setdefault(key, set()).update(values) + + st.session_state["features"] = { + "flows": all_flows, + "artifacts": {key: uniq_sorted(values) for key, values in artifact_values.items()}, + } + st.session_state["dash_aggregates"] = compute_flow_aggregates(all_flows, top_n=10, weight="flows") + st.session_state["zeek_tables"] = batch_result.merged_zeek + st.session_state["osint"] = batch_result.merged_osint + st.session_state["beacon_df"] = batch_result.merged_beacons + st.session_state["dns_analysis"] = batch_result.aggregated_dns + st.session_state["tls_analysis"] = batch_result.aggregated_tls + st.session_state["yara_results"] = _merge_yara_results(analyses) + st.session_state["carved"] = [item for result in batch_result.pcap_results for item in result.carved_items] + st.session_state["__total_pkts"] = batch_result.correlation.total_packets + st.session_state["pipeline_warnings"] = sorted( + {warning for result in batch_result.pcap_results for warning in result.warnings} + ) + st.session_state["pipeline_stages"] = sorted( + {stage for result in batch_result.pcap_results for stage in result.stages_run} + ) + st.session_state["rdns_map"] = { + ip: hostname for result in batch_result.pcap_results for ip, hostname in result.rdns_map.items() + } + reports = [ + f"# {result.filename}\n\n{analysis.report}" + for result, analysis in zip(batch_result.pcap_results, analyses) + if analysis.report + ] + st.session_state["report"] = "\n\n---\n\n".join(reports) or None + st.session_state["llm_status"] = "generated" if reports else None + st.session_state["__pcap_path"] = analyses[0].pcap_path + st.session_state["__pcap_paths"] = [analysis.pcap_path for analysis in analyses] + st.session_state["__batch_mode"] = True + st.session_state["__batch_result"] = batch_result + st.session_state["filter_ips"] = set() + st.session_state["filter_protos"] = set() + st.session_state["filter_time"] = None + + _restore_expensive_derived_state(analyses) + try: + st.session_state["attack_mapping"] = build_attack_mapping(st.session_state) + st.session_state["capture_metrics"] = build_capture_metrics(st.session_state) + except Exception as exc: + logger.warning("Could not rebuild restored ATT&CK state: %s", exc) + st.session_state["attack_mapping"] = None + st.session_state["capture_metrics"] = build_capture_metrics(st.session_state) + st.session_state["restored_analysis_id"] = max( + analyses, key=lambda analysis: analysis.analyzed_at or datetime.min + ).id + st.session_state["restored_analysis_ids"] = [analysis.id for analysis in analyses] + + def render_cases_tab(): """Main cases tab with list and detail views.""" st.markdown("### Case Management") @@ -617,6 +819,7 @@ def _quick_save_analysis(): tls_analysis=st.session_state.get("tls_analysis"), attack_mapping=attack_mapping, capture_metrics=st.session_state.get("capture_metrics"), + session_artifacts=_current_session_artifacts(), ) # Extract IOCs @@ -655,6 +858,7 @@ def _add_current_analysis_to_case(case: Case): tls_analysis=st.session_state.get("tls_analysis"), attack_mapping=attack_mapping, capture_metrics=st.session_state.get("capture_metrics"), + session_artifacts=_current_session_artifacts(), ) analysis.iocs = repo.extract_iocs(analysis) diff --git a/docs/en/USER_MANUAL.md b/docs/en/USER_MANUAL.md index c0a68e3..041b557 100644 --- a/docs/en/USER_MANUAL.md +++ b/docs/en/USER_MANUAL.md @@ -78,6 +78,11 @@ Open the **Upload** tab. Click **Extract & Analyze** to start. +The run is immediately queued as an autosaved Case. Analysis continues in a +separate worker process if you press Streamlit's upper-right **Stop** control or +reload the browser. Reopening the app within seven days reattaches the most +recent UI job; completed evidence can always be reopened from **Cases**. + --- ## The Analysis Pipeline & Progress Tab @@ -97,13 +102,14 @@ PCAP Hunter runs a 10-stage pipeline: | 9 | OSINT Enrichment | Multi-provider reputation lookups | | 10 | LLM Report Generation | AI threat synthesis | -**Execution shape:** stages 2–3 (PyShark, Zeek) run **in parallel**; once both finish, stages 4–7 (DNS, TLS, beaconing, carving) fan out **concurrently**. The Progress tab shows this live: +**Execution shape:** stages 2–3 (PyShark, Zeek) run **in parallel**; once both finish, stages 4–7 (DNS, TLS, beaconing, carving) fan out **concurrently**. The Progress tab monitors the durable job: -- Each phase has its own progress bar and a **caption** describing what is currently happening. -- Every phase has a **Skip** button — long Zeek run on a huge capture? Skip it and the pipeline continues. Panels that depended on the skipped stage will say so explicitly (see [Dashboard](#dashboard) empty states) instead of rendering blank. -- The final phase, **LLM Report Analysis**, generates the AI report; its progress is tracked like any other stage. +- The overall progress bar and per-file rows identify the active stage and job status. +- Streamlit's upper-right **Stop** control pauses only the page display; it does not cancel the worker or erase completed stages. +- Pipeline components can be enabled or disabled in **Config** before submission, and LLM report generation has its own checkbox on **Upload**. +- The final phase, **LLM Report Analysis**, runs in the worker and its report is persisted with the analysis. -You can switch to the Dashboard as soon as the pipeline finishes; partial results appear per-stage. +When the pipeline finishes, the saved results are restored into Dashboard, MITRE Analysis, LLM Analysis, OSINT, and Raw Data. --- diff --git a/docs/zh-TW/README.md b/docs/zh-TW/README.md index de744ff..9764547 100644 --- a/docs/zh-TW/README.md +++ b/docs/zh-TW/README.md @@ -19,6 +19,7 @@ - **獨立的 MITRE ATT&CK 工作區** — 以證據為本的技術假設、ATT&CK v19.1 中繼資料、分析師處置、擷取涵蓋範圍、可視性缺口與 Navigator 匯出。 - **擷取品質遙測** — 封包/流量規模、解析比率、時間範圍、取樣上限、完成階段與警告會隨 UI/API 結果傳遞，並與案件一同保存。 +- **可復原的 UI 分析** — Streamlit 會把 PCAP 工作提交至獨立行程佇列，將完整證據自動保存至 SQLite，並在停止頁面或重新載入瀏覽器後復原近期工作。 - **更安全的 PCAP 攝取** — Streamlit 上傳採有限區塊串流，保留 `.pcap`/`.pcapng`，驗證 Magic Bytes、執行批次限制，拒絕時清除不完整檔案。 - **更強的整合 API** — 無介面工作會回傳 ATT&CK 對應與擷取指標，IOC 摘要包含相關技術 ID，失敗提交會清除暫存案件與檔案。 - **不依賴 LLM 的證據檢視** — 即使跳過或無法產生 AI 敘事，確定性的封包、IOC、關聯、階段與警告證據仍然可見。 @@ -55,7 +56,7 @@ ### 2. Progress — 透明的 10 階段管道 -每個階段都即時回報進度，並提供逐階段的跳過控制。PyShark 與 Zeek 平行執行，接著 DNS、TLS、信標偵測與酬載提取同時展開——你永遠知道目前正在執行什麼、還剩多少進度。 +每個階段都會回報可持久化的工作進度。PyShark 與 Zeek 平行執行，接著 DNS、TLS、信標偵測與酬載提取同時展開。分析在 Streamlit 頁面執行緒之外運行，因此右上角的 Stop 控制或瀏覽器重新載入都不會丟失工作；完成的證據會自動保存至 Cases 並復原。 ![Progress 分頁](../images/02-progress.png) @@ -397,7 +398,7 @@ make run # 獨立安裝（先執行 python3 scripts/install.py） 1. **上傳** — 在 Upload 分頁拖放一個或多個 `.pcap` 檔案。多個檔案會啟動批次模式並進行跨檔案關聯分析。 2. **設定** — 在 Config 分頁選擇 LLM 供應商（LM Studio / OpenAI / Anthropic）、設定自家位置（洲 > 國家 > 城市）、OSINT API 金鑰，並可選擇性指定 YARA 規則目錄。 3. **分析** — 點擊 **Extract & Analyze** 啟動管道。 -4. **監控** — 在 Progress 分頁觀察各階段執行：封包計數 > 解析 + Zeek（平行）> DNS / TLS / 信標偵測 / 酬載提取（同時執行）> YARA > OSINT > LLM 報告。 +4. **監控** — 在 Progress 分頁觀察獨立背景行程中的各階段：封包計數 > 解析 + Zeek（平行）> DNS / TLS / 信標偵測 / 酬載提取（同時執行）> YARA > OSINT > LLM 報告。停止或重新載入 Streamlit 頁面不會丟失工作。 5. **審閱** — 在 Dashboard、MITRE Analysis、LLM Analysis、OSINT、Raw Data、Cases 分頁瀏覽結果。 6. **匯出** — 下載 CSV/JSON 資料、PDF 報告、STIX 套件、ATT&CK Navigator 圖層或 CEF syslog 事件。 diff --git a/docs/zh-TW/USER_MANUAL.md b/docs/zh-TW/USER_MANUAL.md index 6baa5e3..89da301 100644 --- a/docs/zh-TW/USER_MANUAL.md +++ b/docs/zh-TW/USER_MANUAL.md @@ -78,6 +78,8 @@ make run # → http://localhost:8501 點擊 **Extract & Analyze** 開始分析。 +工作會立刻排入佇列並建立自動保存的 Case。即使按下 Streamlit 右上角的 **Stop** 或重新載入瀏覽器，分析仍會在獨立工作行程中繼續。七天內重新開啟應用程式會自動接回最近的 UI 工作；已完成的證據也可隨時從 **Cases** 重新開啟。 + --- ## 分析管道與 Progress 分頁 @@ -97,13 +99,14 @@ PCAP Hunter 執行 10 階段管道： | 9 | OSINT 豐富化 | 多供應商信譽查詢 | | 10 | LLM 報告產生 | AI 威脅綜整 | -**執行形態：** 階段 2–3（PyShark、Zeek）**平行執行**；兩者完成後，階段 4–7（DNS、TLS、信標偵測、酬載提取）**同時**展開。Progress 分頁即時呈現這一切： +**執行形態：** 階段 2–3（PyShark、Zeek）**平行執行**；兩者完成後，階段 4–7（DNS、TLS、信標偵測、酬載提取）**同時**展開。Progress 分頁會監看可復原的背景工作： -- 每個階段都有自己的進度條與**說明文字（caption）**，描述目前正在進行的工作。 -- 每個階段都有 **Skip** 按鈕——大型擷取檔讓 Zeek 跑太久？跳過它，管道會繼續執行。依賴被跳過階段的面板會明確說明（見[儀表板](#儀表板)的空狀態），而不是呈現空白。 -- 最後一個階段 **LLM Report Analysis** 負責產生 AI 報告，其進度與其他階段相同方式追蹤。 +- 整體進度條與逐檔案列會顯示目前階段及工作狀態。 +- Streamlit 右上角的 **Stop** 只會暫停頁面顯示，不會取消工作行程或刪除已完成階段。 +- 提交前可在 **Config** 啟用或停用管道元件；**Upload** 另有是否產生 LLM 報告的核取方塊。 +- 最後一個階段 **LLM Report Analysis** 會在工作行程中執行，報告也會與分析一同保存。 -管道完成後即可切換到儀表板；各階段的結果會分別呈現。 +管道完成後，保存的結果會復原至 Dashboard、MITRE Analysis、LLM Analysis、OSINT 與 Raw Data。 --- diff --git a/requirements.txt b/requirements.txt index ce08bab..b6898a3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ # Core app dependencies -streamlit>=1.36,<2.0 +streamlit>=1.37,<2.0 pandas>=2.2,<3.0 numpy>=1.26,<3.0 matplotlib>=3.9,<4.0 diff --git a/tests/api/test_queue.py b/tests/api/test_queue.py index 75de3da..0b3c258 100644 --- a/tests/api/test_queue.py +++ b/tests/api/test_queue.py @@ -23,7 +23,7 @@ cancel_queued_job, recover_stale_running_jobs, ) -from app.database.models import Case, CaseStatus, Job, JobStatus, Severity +from app.database.models import Analysis, Case, CaseStatus, Job, JobStatus, Severity from app.database.repository import CaseRepository @@ -287,6 +287,93 @@ def fake_run_pipeline(pcap_path, case_id, options, progress, heartbeat=None): assert result["capture_metrics"]["detectors"]["zeek"] == "available" persisted = repo.get_analysis(result["analysis_id"]) assert persisted.features["beacon_records"] == records + assert persisted.session_artifacts["beacon_records"] == records + assert persisted.session_artifacts["zeek_tables"] == {"conn": [{"uid": "x"}]} + + +def test_worker_llm_opt_in_persists_report(tmp_path, monkeypatch): + """Streamlit background jobs keep the LLM result outside Session State.""" + import app.llm.providers as provider_mod + import app.pipeline.runner as runner_mod + from app.pipeline.runner import PipelineResult + + def fake_run_pipeline(pcap_path, case_id, options, progress, heartbeat=None): + return PipelineResult( + case_id=case_id, + packet_count=1, + features={ + "flows": [], + "artifacts": {"ips": [], "domains": [], "urls": [], "hashes": [], "ja3": []}, + }, + ) + + monkeypatch.setattr(runner_mod, "run_pipeline", fake_run_pipeline) + monkeypatch.setattr( + queue_mod, + "_load_llm_settings", + lambda: ("lmstudio", "http://localhost:1234/v1", "", "test-model", "US English"), + ) + monkeypatch.setattr(provider_mod, "synthesize_report", lambda *args, **kwargs: "# Durable report") + + fake_pcap = tmp_path / "fake.pcap" + fake_pcap.write_bytes(b"\xd4\xc3\xb2\xa1" + b"\x00" * 20) + db = str(tmp_path / "t.db") + repo = CaseRepository(db_path=db) + repo.create_case(Case(id="cafe0007", title="t", status=CaseStatus.IN_PROGRESS, severity=Severity.LOW)) + job_id = repo.create_job(Job(case_id="cafe0007", pcap_path=str(fake_pcap), options_json="{}")) + + _worker_run( + job_id, + db, + str(fake_pcap), + {"osint_enabled": False, "llm_enabled": True, "do_yara": False}, + ) + + result = json.loads(repo.get_job(job_id).result_json) + persisted = repo.get_analysis(result["analysis_id"]) + assert result["summary_narrative"] == "# Durable report" + assert persisted.report == "# Durable report" + assert "llm" in persisted.session_artifacts["pipeline_stages"] + + +def test_report_only_job_updates_existing_analysis(tmp_path, monkeypatch): + import app.llm.providers as provider_mod + + monkeypatch.setattr( + queue_mod, + "_load_llm_settings", + lambda: ("lmstudio", "http://localhost:1234/v1", "", "test-model", "US English"), + ) + monkeypatch.setattr(provider_mod, "synthesize_report", lambda *args, **kwargs: "# Updated report") + + fake_pcap = tmp_path / "fake.pcap" + fake_pcap.write_bytes(b"\xd4\xc3\xb2\xa1" + b"\x00" * 20) + db = str(tmp_path / "t.db") + repo = CaseRepository(db_path=db) + case_id = repo.create_case(Case(title="report-only", status=CaseStatus.IN_PROGRESS)) + analysis = Analysis( + case_id=case_id, + pcap_path=str(fake_pcap), + features={"flows": [], "artifacts": {"ips": []}}, + report="# Old report", + session_artifacts={"pipeline_stages": ["pyshark_pass"]}, + ) + analysis_id = repo.save_analysis(analysis) + options = { + "_job_type": "llm_report", + "_analysis_id": analysis_id, + "llm_enabled": True, + } + job_id = repo.create_job(Job(case_id=case_id, pcap_path=str(fake_pcap), options_json=json.dumps(options))) + + _worker_run(job_id, db, str(fake_pcap), options) + + result = json.loads(repo.get_job(job_id).result_json) + persisted = repo.get_analysis(analysis_id) + assert result["analysis_id"] == analysis_id + assert persisted.report == "# Updated report" + assert persisted.features == {"flows": [], "artifacts": {"ips": []}} + assert repo.get_case(case_id).analysis_count == 1 # --------------------------------------------------------------------------- diff --git a/tests/test_background_analysis.py b/tests/test_background_analysis.py new file mode 100644 index 0000000..f1071d1 --- /dev/null +++ b/tests/test_background_analysis.py @@ -0,0 +1,124 @@ +"""Durable Streamlit background-job submission, recovery, and restore tests.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock + +from app.database.models import Analysis, Case, Job, JobStatus +from app.database.repository import CaseRepository +from app.ui import background_analysis as bg + + +class _RecordingQueue: + def __init__(self, repo: CaseRepository): + self.repo = repo + self.submissions = [] + + def enqueue(self, submission): + self.submissions.append(submission) + return self.repo.create_job( + Job( + case_id=submission.case_id, + pcap_path=submission.pcap_path, + options_json=json.dumps(submission.options), + ) + ) + + +def test_submit_creates_autosaved_case_and_durable_jobs(tmp_path, monkeypatch): + repo = CaseRepository(db_path=str(tmp_path / "cases.db")) + queue = _RecordingQueue(repo) + monkeypatch.setattr(bg, "get_background_repo", lambda: repo) + monkeypatch.setattr(bg, "get_background_queue", lambda: queue) + + run = bg.submit_background_analysis( + ["/tmp/one.pcap", "/tmp/two.pcap"], + {"do_zeek": True, "llm_enabled": False}, + ) + + case = repo.get_case(run["case_id"]) + assert case is not None + assert case.status.value == "in_progress" + assert case.tags == ["autosaved", "background"] + assert len(run["job_ids"]) == 2 + assert queue.submissions[0].options["_origin"] == "streamlit" + assert queue.submissions[1].options["_batch_index"] == 1 + + +def test_recovery_prefers_active_streamlit_run(tmp_path, monkeypatch): + repo = CaseRepository(db_path=str(tmp_path / "cases.db")) + old_case = repo.create_case(Case(title="old")) + active_case = repo.create_case(Case(title="active")) + old_id = repo.create_job( + Job(case_id=old_case, pcap_path="/tmp/old.pcap", options_json=json.dumps({"_origin": "streamlit"})) + ) + repo.update_job_status(old_id, JobStatus.DONE) + active_id = repo.create_job( + Job(case_id=active_case, pcap_path="/tmp/new.pcap", options_json=json.dumps({"_origin": "streamlit"})) + ) + repo.update_job_status(active_id, JobStatus.RUNNING) + # A non-Streamlit API job must never be attached to the UI session. + repo.create_job(Job(case_id=active_case, pcap_path="/tmp/api.pcap", options_json="{}")) + monkeypatch.setattr(bg, "get_background_repo", lambda: repo) + + recovered = bg.find_recoverable_background_run() + + assert recovered["case_id"] == active_case + assert recovered["job_ids"] == [active_id] + assert recovered["recovered"] is True + + +def test_submit_report_reuses_case_and_analysis(tmp_path, monkeypatch): + repo = CaseRepository(db_path=str(tmp_path / "cases.db")) + queue = _RecordingQueue(repo) + case_id = repo.create_case(Case(title="report")) + analysis = Analysis(case_id=case_id, pcap_path="/tmp/report.pcap", features={"flows": []}) + analysis_id = repo.save_analysis(analysis) + monkeypatch.setattr(bg, "get_background_repo", lambda: repo) + monkeypatch.setattr(bg, "get_background_queue", lambda: queue) + + run = bg.submit_background_report([analysis_id]) + + assert run["case_id"] == case_id + assert run["report_only"] is True + assert queue.submissions[0].options["_job_type"] == "llm_report" + assert queue.submissions[0].options["_analysis_id"] == analysis_id + assert repo.get_case(case_id).status.value == "in_progress" + + +def test_completed_job_restores_persisted_analysis(tmp_path, monkeypatch): + repo = CaseRepository(db_path=str(tmp_path / "cases.db")) + case_id = repo.create_case(Case(title="complete")) + analysis = Analysis( + case_id=case_id, + pcap_path="/tmp/complete.pcap", + features={"flows": [], "artifacts": {}}, + session_artifacts={"pipeline_stages": ["pyshark_pass"]}, + ) + analysis_id = repo.save_analysis(analysis) + job_id = repo.create_job( + Job( + case_id=case_id, + pcap_path=analysis.pcap_path, + options_json=json.dumps({"_origin": "streamlit"}), + ) + ) + repo.complete_job(job_id, json.dumps({"analysis_id": analysis_id}).encode()) + + fake_st = MagicMock() + fake_st.session_state = {} + restore = MagicMock() + monkeypatch.setattr(bg, "get_background_repo", lambda: repo) + monkeypatch.setattr(bg, "restore_analyses_to_session", restore) + monkeypatch.setattr(bg, "st", fake_st) + run = {"case_id": case_id, "job_ids": [job_id], "loaded": False} + + restored = bg.render_background_progress(run) + + assert restored is True + restore.assert_called_once() + assert restore.call_args.args[0][0].id == analysis_id + assert run["loaded"] is True + assert fake_st.session_state[bg.BACKGROUND_RUN_KEY]["loaded"] is True + assert repo.get_case(case_id).status.value == "open" diff --git a/tests/test_case_management.py b/tests/test_case_management.py index 7ceed55..551e134 100644 --- a/tests/test_case_management.py +++ b/tests/test_case_management.py @@ -171,12 +171,14 @@ def test_to_dict_and_repository_round_trip_new_analytic_fields(self, tmp_path): pcap_path="/tmp/test.pcap", attack_mapping={"attack_version": "19.1", "techniques": [{"technique_id": "T1571"}]}, capture_metrics={"flow_count": 4, "visibility_gaps": ["zeek"]}, + session_artifacts={"pipeline_stages": ["zeek"], "carved": [{"sha256": "abc"}]}, ) payload = analysis.to_dict() restored = Analysis.from_dict(payload) assert restored.attack_mapping["attack_version"] == "19.1" assert restored.capture_metrics["flow_count"] == 4 + assert restored.session_artifacts["pipeline_stages"] == ["zeek"] repo = CaseRepository(db_path=str(tmp_path / "cases.db")) analysis_id = repo.save_analysis(analysis) @@ -185,6 +187,7 @@ def test_to_dict_and_repository_round_trip_new_analytic_fields(self, tmp_path): assert persisted is not None assert persisted.attack_mapping["techniques"][0]["technique_id"] == "T1571" assert persisted.capture_metrics["visibility_gaps"] == ["zeek"] + assert persisted.session_artifacts["carved"] == [{"sha256": "abc"}] def test_from_dict(self): data = { diff --git a/tests/test_case_restore.py b/tests/test_case_restore.py index 001d0fb..eb94923 100644 --- a/tests/test_case_restore.py +++ b/tests/test_case_restore.py @@ -117,6 +117,31 @@ def test_stale_non_persisted_keys_reset(self): assert st.session_state["port_anomalies"] is None assert st.session_state["rdns_map"] == {} + def test_durable_background_artifacts_restored(self): + analysis = _make_analysis( + features={ + "flows": [], + "artifacts": {"ips": [], "domains": [], "urls": [], "hashes": [], "ja3": []}, + }, + session_artifacts={ + "beacon_records": [{"src": "10.0.0.1", "dst": "8.8.8.8", "score": 0.9}], + "zeek_tables": {"dns.log": [{"query": "durable.example"}]}, + "carved": [{"sha256": "abc123", "path": "/tmp/carved.bin"}], + "pipeline_warnings": ["zeek_no_logs"], + "pipeline_stages": ["pyshark_pass", "carve"], + "rdns_map": {"8.8.8.8": "dns.google"}, + }, + ) + + _restore_analysis_to_session(analysis) + + assert st.session_state["beacon_df"].iloc[0]["score"] == 0.9 + assert st.session_state["zeek_tables"]["dns.log"].iloc[0]["query"] == "durable.example" + assert st.session_state["carved"][0]["sha256"] == "abc123" + assert st.session_state["pipeline_warnings"] == ["zeek_no_logs"] + assert st.session_state["pipeline_stages"] == ["pyshark_pass", "carve"] + assert st.session_state["rdns_map"] == {"8.8.8.8": "dns.google"} + def test_stale_dashboard_filters_cleared(self): """Filters reference IPs/time ranges from the prior capture; leaving them active would keep filtered_flows empty — the exact symptom being fixed.""" diff --git a/tests/test_job_repository.py b/tests/test_job_repository.py index 42ddb9b..9e9fdbf 100644 --- a/tests/test_job_repository.py +++ b/tests/test_job_repository.py @@ -120,3 +120,17 @@ def test_count_active_jobs(tmp_path): conn.close() repo.update_job_status(row["id"], JobStatus.DONE) assert repo.count_active_jobs() == 1 + + +def test_list_jobs_filters_by_case_and_status(tmp_path): + repo = _setup_repo(tmp_path) + case_a = repo.create_case(Case(title="A")) + case_b = repo.create_case(Case(title="B")) + queued_a = repo.create_job(Job(case_id=case_a, pcap_path="/tmp/a.pcap")) + done_a = repo.create_job(Job(case_id=case_a, pcap_path="/tmp/b.pcap")) + repo.create_job(Job(case_id=case_b, pcap_path="/tmp/c.pcap")) + repo.update_job_status(done_a, JobStatus.DONE) + + jobs = repo.list_jobs(case_id=case_a, statuses=[JobStatus.QUEUED]) + + assert [job.id for job in jobs] == [queued_a] diff --git a/tests/test_jobs_schema.py b/tests/test_jobs_schema.py index b9859e5..19a7f2d 100644 --- a/tests/test_jobs_schema.py +++ b/tests/test_jobs_schema.py @@ -52,3 +52,13 @@ def test_jobs_table_columns(tmp_path): "result_json", } assert expected.issubset(cols), f"Missing columns: {expected - cols}" + + +def test_analyses_has_session_artifacts_column(tmp_path): + repo = CaseRepository(db_path=str(tmp_path / "test.db")) + conn = repo._get_conn() + try: + cols = {r[1] for r in conn.execute("PRAGMA table_info(analyses)").fetchall()} + finally: + conn.close() + assert "session_artifacts_json" in cols From c13118c3fee044cade43e9aa22c69fae46157949 Mon Sep 17 00:00:00 2001 From: Henry Hu Date: Tue, 14 Jul 2026 18:06:58 +0700 Subject: [PATCH 2/2] chore: remove internal tooling and planning docs --- .dockerignore | 1 - .gitignore | 6 - CHANGELOG.md | 2 +- CLAUDE.md | 235 --------- README.md | 2 - docs/FEATURE-ROADMAP.md | 498 ------------------- docs/PHASE4-IMPLEMENTATION-PLAN.md | 736 ---------------------------- docs/TEST-PLAN.md | 752 ----------------------------- docs/zh-TW/README.md | 2 - 9 files changed, 1 insertion(+), 2233 deletions(-) delete mode 100644 CLAUDE.md delete mode 100644 docs/FEATURE-ROADMAP.md delete mode 100644 docs/PHASE4-IMPLEMENTATION-PLAN.md delete mode 100644 docs/TEST-PLAN.md diff --git a/.dockerignore b/.dockerignore index e2b45d7..4a8b2e2 100644 --- a/.dockerignore +++ b/.dockerignore @@ -9,7 +9,6 @@ htmlcov/ __pycache__/ **/__pycache__/ *.pyc -.claude/ .dockerignore Dockerfile docker-compose.yml diff --git a/.gitignore b/.gitignore index 3dc1c01..f577cfd 100644 --- a/.gitignore +++ b/.gitignore @@ -44,11 +44,5 @@ data/ *.log .DS_Store -# Claude Code local state (plans, worktrees, agents, memory) -# CLAUDE.md is intentionally tracked (documents project conventions); -# everything under .claude/ is session-local and must not be committed. -.claude/ -CLAUDE.local.md - # Superpowers auto-generated specs and plans (session artifacts, not source) docs/superpowers/ diff --git a/CHANGELOG.md b/CHANGELOG.md index f947bad..70b8ebe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,7 +57,7 @@ First stable release. Production-ready installer, hardened pipeline, polished UX ### Changed - **PDF cover page redesign** — logo + tagline above the title, with classification banner and metadata block. - **Kaleido upgraded to 1.x** — 0.x reaches end-of-life September 2025; 1.x is the active branch. -- **Testing discipline overhauled** — production-shape test data (real `CorrelationSignal` dataclasses, real DataFrames, nested dicts the pipeline actually produces) instead of simplified inputs. Documented in `CLAUDE.md` with bug-pattern history. New integration tests for every PDF section and chart. +- **Testing discipline overhauled** — production-shape test data (real `CorrelationSignal` dataclasses, real DataFrames, nested dicts the pipeline actually produces) instead of simplified inputs. New integration tests cover every PDF section and chart. - **Version bumped to 1.0.0** with consolidated release notes. ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index 2cab302..0000000 --- a/CLAUDE.md +++ /dev/null @@ -1,235 +0,0 @@ -# CLAUDE.md — PCAP Hunter - -## Project Overview - -PCAP Hunter is an AI-enhanced threat hunting workbench for SOC analysts. It combines network analysis tools (Zeek, Tshark, PyShark) with LLMs and OSINT APIs to ingest, analyze, and extract actionable intelligence from PCAP files. Built with Streamlit for the web UI. - -## Quick Reference - -```bash -make install # pip install -r requirements.txt -make run # streamlit run app/main.py (port 8501) -make test # PYTHONPATH=. pytest tests/ -v --cov=app -make lint # ruff check . -make format # ruff format . -make clean # Remove caches -make docker-up # Build + run the UI in Docker (http://localhost:8501) -make docker-verify # Format + lint + full test suite INSIDE the image -``` - -Always run tests with `PYTHONPATH=.` — this is required for absolute imports to resolve. - -### Docker is the canonical build-and-verify path - -**Any local verification that requires a build must go through Docker** -(`make docker-up` to run the app, `make docker-verify` for the gate). The -host machine has multiple coexisting Python installs (framework + Homebrew), -so host-side "it works here" proves nothing about a clean environment — -a fresh-install breakage shipped exactly that way once. `make verify` on the -host remains fine for fast iteration; anything build-shaped (dependency -changes, install paths, release checks, user-facing verification) runs in -the container. - -API keys saved inside the container persist in the `pcap-hunter-home` volume -(mounted at `/home/runner`, with a pinned `hostname:` so the config encryption -key stays stable); LM Studio on the host is reachable via -`host.docker.internal` — the compose file defaults `LM_BASE_URL` accordingly. - -## Architecture - -``` -app/ -├── main.py # Streamlit entry point (session state, 9-tab UI) -├── config.py # App defaults & constants (thresholds, paths) -├── analysis/ # Scoring, correlation, flow analysis, narration -├── database/ # SQLite case management (models.py, repository.py) -├── llm/ # OpenAI-compatible API client (LM Studio supported) -├── pipeline/ # 10-stage analysis pipeline (pcap → report) -├── reports/ # PDF generation (WeasyPrint + Jinja2) -├── security/ # OPSEC hardening (sanitization, secure HTTP) -├── threat_intel/ # MITRE ATT&CK mapping engine -├── ui/ # Streamlit components (layout, charts, config, cases) -└── utils/ # Shared utilities (export, crypto, network, YARA) -tests/ # 22+ test modules, one per major component -docs/ # EN + zh-TW user manuals, roadmap, test plan -data/ # Runtime artifacts (carved/, zeek/, *.db) — gitignored -``` - -### 10-Stage Pipeline - -1. `pcap_count.py` — Fast packet counting (tshark) -2. `pyshark_pass.py` — Deep packet parsing (up to 200K packets) -3. `zeek.py` — Automated Zeek execution and log parsing -4. `dns_analysis.py` — DGA detection, DNS tunneling, fast flux -5. `tls_certs.py` — TLS/SSL certificate chain validation -6. `beacon.py` — C2 beaconing detection (statistical analysis) -7. `carve.py` — HTTP payload extraction with SHA256 hashing -8. `yara_scan.py` — YARA rule-based file scanning -9. `osint.py` — Multi-provider OSINT enrichment (VT, AbuseIPDB, Shodan, etc.) -10. LLM synthesis — AI-powered threat report generation - -Stages 2–3 (PyShark, Zeek) run in parallel; after that parse join, stages 4–7 -(DNS, TLS, beaconing, carving) run concurrently. Zeek and carve write into -per-run output dirs (`data/zeek|carved/_/`) so concurrent runs -never clobber each other; stale run dirs are pruned after 7 days. - -## Tech Stack - -- **Python 3.11+**, Streamlit 1.36+, Pandas 2.2+, NumPy 1.26+ -- **Network**: Zeek, Tshark, PyShark 0.6+, Scapy 2.5+ -- **LLM**: OpenAI SDK 1.30+ (LM Studio compatible) -- **OSINT**: VirusTotal, AbuseIPDB, GreyNoise, OTX, Shodan, MaxMind GeoIP -- **Security**: cryptography 42.0+ (PBKDF2 encrypted config) -- **Export**: WeasyPrint (PDF), STIX 2.0/2.1, ATT&CK Navigator, CSV/JSON -- **Dev**: Ruff (lint+format), Pytest 8.0+, pytest-cov, GitHub Actions CI - -## Code Conventions - -### Style - -- **Formatter/Linter**: Ruff — line length 120, double quotes, 4-space indent -- **Lint rules**: E, F, I, W (see `pyproject.toml` for per-file ignores) -- **Type hints**: Used extensively; `from __future__ import annotations` for forward compat -- **Docstrings**: Google-style (Args/Returns sections) on public functions - -### Naming - -- Modules: `snake_case.py` (e.g., `ioc_scorer.py`, `dns_analysis.py`) -- Classes: `PascalCase` (e.g., `IOCScorer`, `ConfigManager`) -- Functions: `snake_case` (e.g., `rank_beaconing`, `validate_domain`) -- Constants: `UPPER_SNAKE_CASE` (e.g., `MAX_DOMAIN_LENGTH`, `DATA_DIR`) -- Private: leading underscore (e.g., `_sanitize_for_llm`) - -### Imports - -- Always use **absolute imports**: `from app.pipeline.beacon import rank_beaconing` -- Standard library → third-party → local app modules (enforced by ruff `I` rule) -- Backward-compatible re-exports exist in `app/utils/common.py` - -### Error Handling - -- Custom exceptions inherit from `Exception` (e.g., `CarveError`) -- Use `logging` module, not print statements -- Streamlit phases use `phase.done()` for completion tracking - -### Data Modeling - -- Prefer `dataclass` for structured data -- Enums for fixed categories (e.g., `Severity`, `CaseStatus`) - -## Testing - -- **Framework**: Pytest with `--cov=app` -- **Location**: `tests/test_.py` — one test file per major module -- **Run**: `make test` or `PYTHONPATH=. pytest tests/ -v --cov=app` -- **Pre-commit gate**: `make verify` — runs format check + lint + full test suite -- **PDF-focused**: `make test-pdf` — PDF generator, chart images, chart rendering, integration -- **Conventions**: - - Test classes: `Test` (e.g., `TestTechniqueMatch`) - - Test functions: `test_` (e.g., `test_periodicity_score_empty`) - - Test both happy path and edge cases (empty, None, malformed) - - Use dataclass instances for complex test objects - - No shared conftest.py fixtures — tests are independent - -### Testing discipline (non-negotiable) - -**Before every commit, run `make verify`.** It must pass. CI runs the same -checks, so if `make verify` fails locally it will fail in CI. - -**Use production-shape test data, not "looks reasonable" dicts.** If the -code consumes `list[CorrelationSignal]` dataclasses, tests must pass actual -dataclasses — not dicts that happen to have similar keys. Simplified -inputs are exactly what let the `', '.join(c.signals)` bug slip past 500+ -unit tests into production. See `tests/test_pdf_integration.py` for the -shapes every PDF-related code path should accept. - -**When adding a new PDF section, extend the integration test.** A new -`_render_*_section()` in `pdf_generator.py` needs a corresponding assertion -in `test_pdf_integration.py::test_html_contains_every_expected_section` -that verifies the section ID and at least one key token appear in the -output HTML. - -**When adding a new chart to the PDF, extend `test_chart_rendering.py`.** -Any chart called from `_render_charts_section` must have a smoke test that -passes production-shape data and asserts the figure renders to PNG via -kaleido. This catches two recurring bug classes: - 1. API drift between the chart function and the PDF call site - 2. Runtime render failures (colorscale issues, unicode, etc.) - -**`@pytest.mark.skipif` on macOS — import `pdf_generator` first.** Any -test that skips based on `WEASYPRINT_AVAILABLE` must import `pdf_generator` -**before** touching `weasyprint`, otherwise the dyld path fix hasn't run -yet and the test silently skips on a working system. See the import block -in `tests/test_pdf_generator.py` for the correct pattern. - -### Historical bug patterns — don't repeat these - -| Bug | Lesson | -|-----|--------| -| LLM sections showed duplicated headings (`## Title\n\nTitle\n\n...`) | Don't wrap section names in `**bold**` in prompts — LLMs echo them back. Strip leading title lines as a safety net. | -| WeasyPrint crashed with `OSError: cannot load library 'libgobject-2.0-0'` | macOS dyld doesn't search `/opt/homebrew/lib` by default. Set `DYLD_FALLBACK_LIBRARY_PATH` before the `weasyprint` import. Catch `(ImportError, OSError)`, not just `ImportError`. | -| PDF generation crashed with `TypeError: expected str instance, CorrelationSignal found` | `c.signals` is `list[CorrelationSignal]`, not `list[str]`. Extract `.name` before joining. | -| PDF tests silently skipped on macOS despite working app | Test module imported `weasyprint` directly before `pdf_generator`, so DYLD fix hadn't run. Always import `pdf_generator` first. | -| Empty dashboard after PCAP upload | `tshark` wasn't installed. Pre-flight check in `app/main.py` now shows a red banner. | -| Chart API mismatch between function and call site | `plot_top_n_charts` expects a flat `dict[str, int]`, not nested. `plot_network_graph` takes `flows` positionally. Smoke tests in `test_chart_rendering.py` catch this. | -| Header icon looked "cut off" on the dark theme | Two PIL bugs in `scripts/build_logo_assets.py`: `paste(im, box, mask)` *replaces* pixels (erased the lens disc with transparency) — composite via `putalpha(multiply)` + `alpha_composite` instead; and `ellipse(outline=, width=)` leaves moiré gaps at large widths — build rings from two filled circles. Logo is theme-aware (`resolve_logo_path` + `st.context.theme`); `tests/test_logo_assets.py` guards the generated PNGs. | - -## CI/CD - -GitHub Actions (`.github/workflows/ci.yml`): -- Triggers on push/PR to `main` -- Python 3.11, Ubuntu latest -- Steps: install deps → pytest with coverage → ruff check → ruff format --check - -## Configuration - -- Defaults in `app/config.py` (thresholds, paths, URLs) -- Persistent config: `~/.pcap_hunter_config.json` (via `ConfigManager`) -- API keys encrypted with machine-derived PBKDF2 key -- Environment variable overrides: `OTT_KEY`, `VT_KEY`, `SHODAN_KEY`, etc. -- LLM defaults: `http://localhost:1234/v1` (LM Studio) - -## Key Thresholds (config.py) - -- DGA entropy: 4.0 bits -- Fast flux: domain resolving to 10+ IPs -- Flow asymmetry: 10:1 outbound/inbound ratio, 1MB minimum -- C2 common ports: {4444, 5555, 6666, 7777, 8888, 9999, 1337, 31337} -- Default PyShark limit: 200,000 packets -- OSINT top IPs default: 50 -- `MAX_FLOW_SAMPLES`: 5,000 per-flow packet timestamps/lengths (true totals kept in `count`/`first_ts`/`last_ts`) -- `RUN_DIR_RETENTION_SECONDS`: 7 days — per-run `data/zeek|carved//` dirs pruned on the next run -- Subprocess timeouts: `ZEEK_TIMEOUT_SECONDS` 600, `PCAP_COUNT_TIMEOUT_SECONDS` 120, `CARVE_TIMEOUT_SECONDS` 300, `TLS_EXTRACT_TIMEOUT_SECONDS` 300, `LLM_PROBE_TIMEOUT_SECONDS` 15 - -## Git Conventions - -- Branch: `main` (production) -- Commit messages: conventional-commits style — `feat:`, `fix:`, `docs:`, `style:`, `chore:` -- Lowercase descriptions after prefix -- Always run `make test` and `make lint` before committing - -## Security Considerations - -- Never commit API keys or `.env` files -- Config encryption via `ConfigManager` for sensitive keys -- CSV injection protection in all exports -- Input validation for domains, IPs (ReDoS prevention) -- Hardened HTTP sessions via `opsec.py` -- YARA rules loaded from user-supplied paths — validate carefully - -## Adding a New Pipeline Stage - -1. Create `app/pipeline/.py` with the analysis logic -2. Add phase tracking via `app/pipeline/state.py` -3. Wire into `app/main.py` pipeline execution flow -4. Add corresponding UI rendering in `app/ui/layout.py` -5. Create `tests/test_.py` with edge case coverage -6. Update `app/config.py` if new thresholds/defaults needed - -## Adding a New OSINT Provider - -1. Add API integration in `app/pipeline/osint.py` -2. Add config key in `app/config.py` and `cfg__key` encryption in `ConfigManager` -3. Add UI controls in `app/ui/config_ui.py` -4. Update IOC scoring weights in `app/analysis/ioc_scorer.py` -5. Add tests in `tests/` and update documentation diff --git a/README.md b/README.md index f310a92..5c19e33 100644 --- a/README.md +++ b/README.md @@ -577,8 +577,6 @@ PCAP Hunter uses **production-shape test data**, not simplified inputs. See `tes - **[Integrations API Reference](docs/API.md)** — REST endpoints, authentication, configuration - **[API Integration Guides](docs/api/README.md)** — SIEM / SOAR integration recipes - **[中文說明 (Traditional Chinese README)](docs/zh-TW/README.md)** — 繁體中文版 -- **[CLAUDE.md](CLAUDE.md)** — contributor/AI guide: conventions, testing discipline, known bug patterns -- **[docs/FEATURE-ROADMAP.md](docs/FEATURE-ROADMAP.md)** — planned work --- diff --git a/docs/FEATURE-ROADMAP.md b/docs/FEATURE-ROADMAP.md deleted file mode 100644 index ff41de9..0000000 --- a/docs/FEATURE-ROADMAP.md +++ /dev/null @@ -1,498 +0,0 @@ -# PCAP Hunter Feature Roadmap - -## Overview - -This document outlines the planned features for PCAP Hunter, prioritized by implementation complexity and user value. - ---- - -## Phase 1: Quick Wins (Low Effort, High Value) - -### 1.1 CSV/JSON Export - -**Goal**: Allow users to export analysis results for external processing or archival. - -**Scope**: -- Export flow data, Zeek logs, OSINT results, and beaconing scores -- Support CSV (for spreadsheets) and JSON (for programmatic use) -- Include filtered views (respect current dashboard filters) - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| Export utilities | `app/utils/export.py` | New module | -| UI buttons | `app/ui/results_tab.py` | Add export buttons per section | -| Streamlit download | `app/main.py` | Wire `st.download_button` | - -**Technical Details**: -```python -# app/utils/export.py -def export_to_csv(data: list[dict], filename: str) -> bytes: - """Convert list of dicts to CSV bytes.""" - -def export_to_json(data: Any, filename: str, indent: int = 2) -> bytes: - """Convert data to formatted JSON bytes.""" -``` - -**Dependencies**: None (uses standard library) - ---- - -### 1.2 Configuration Persistence - -**Goal**: Save user settings (API keys, thresholds, preferences) across sessions. - -**Scope**: -- Persist: LLM endpoint, API keys, analysis toggles, threshold values -- Storage: Local JSON file (`.pcap_hunter_config.json`) -- Security: Encrypt sensitive values (API keys) - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| Config manager | `app/utils/config_manager.py` | New module | -| Encryption | `app/security/crypto.py` | New module (Fernet) | -| UI integration | `app/ui/sidebar.py` | Load/save buttons | - -**Technical Details**: -```python -# app/utils/config_manager.py -class ConfigManager: - def load(self) -> dict: ... - def save(self, config: dict) -> None: ... - def get(self, key: str, default: Any = None) -> Any: ... -``` - -**Dependencies**: `cryptography` (for Fernet encryption) - ---- - -### 1.3 OSINT Response Caching - -**Goal**: Cache OSINT API responses to reduce API calls and improve response time. - -**Scope**: -- Cache by IP/domain with configurable TTL (default: 24 hours) -- Storage: SQLite database (`data/osint_cache.db`) -- Manual cache invalidation UI - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| Cache layer | `app/pipeline/osint_cache.py` | New module | -| OSINT integration | `app/pipeline/osint.py` | Check cache before API call | -| UI controls | `app/ui/sidebar.py` | Cache stats & clear button | - -**Technical Details**: -```python -# app/pipeline/osint_cache.py -class OSINTCache: - def __init__(self, db_path: str, ttl_hours: int = 24): ... - def get(self, indicator: str, provider: str) -> dict | None: ... - def set(self, indicator: str, provider: str, data: dict) -> None: ... - def invalidate(self, indicator: str = None) -> int: ... -``` - -**Dependencies**: `sqlite3` (standard library) - ---- - -### 1.4 JA3/JA3S Fingerprint Lookup - -**Goal**: Identify TLS client/server implementations using JA3 fingerprints. - -**Scope**: -- Calculate JA3/JA3S from TLS handshake fields -- Lookup against known fingerprint databases -- Display in SSL/TLS analysis section - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| JA3 calculator | `app/pipeline/ja3.py` | New module | -| Fingerprint DB | `app/data/ja3_fingerprints.json` | Static lookup table | -| Zeek integration | `app/pipeline/zeek.py` | Parse `ssl.log` JA3 fields | -| UI display | `app/ui/results_tab.py` | Add JA3 column to SSL table | - -**Technical Details**: -```python -# app/pipeline/ja3.py -def calculate_ja3(version: str, ciphers: list, extensions: list, - curves: list, point_formats: list) -> str: - """Calculate JA3 fingerprint hash.""" - -def lookup_ja3(ja3_hash: str) -> dict | None: - """Lookup JA3 in known fingerprint database.""" -``` - -**Dependencies**: None (Zeek already extracts JA3) - ---- - -## Phase 2: Medium Effort Features - -### 2.1 DNS Query/Response Carving - -**Goal**: Extract and analyze DNS queries and responses for threat detection. - -**Scope**: -- Parse DNS packets for query names, types, and responses -- Detect suspicious patterns (DGA, DNS tunneling, fast flux) -- Visualize DNS activity timeline - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| DNS carver | `app/pipeline/dns_carve.py` | New module | -| DGA detection | `app/pipeline/dns_analysis.py` | Entropy/pattern analysis | -| Zeek DNS parsing | `app/pipeline/zeek.py` | Enhanced `dns.log` parsing | -| UI visualization | `app/ui/dns_tab.py` | New tab | - -**Technical Details**: -```python -# app/pipeline/dns_analysis.py -def detect_dga(domain: str) -> float: - """Return DGA probability score (0-1).""" - -def detect_tunneling(dns_records: list[dict]) -> dict: - """Analyze for DNS tunneling indicators.""" - -def detect_fast_flux(domain: str, responses: list[dict]) -> bool: - """Check for fast-flux DNS behavior.""" -``` - -**Dependencies**: None (uses existing Zeek/PyShark) - ---- - -### 2.2 Multi-PCAP Batch Analysis - -**Goal**: Analyze multiple PCAP files together with correlation across files. - -**Scope**: -- Upload multiple PCAPs -- Correlate IPs/domains across files -- Timeline aggregation -- Merged reporting - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| Batch processor | `app/pipeline/batch.py` | New module | -| Session merger | `app/pipeline/merge.py` | New module | -| UI multi-upload | `app/ui/upload_tab.py` | Multi-file upload | -| Progress tracking | `app/pipeline/state.py` | Multi-file progress | - -**Technical Details**: -```python -# app/pipeline/batch.py -class BatchProcessor: - def __init__(self, pcap_paths: list[str]): ... - def process_all(self, phase: PhaseHandle) -> dict: ... - def correlate(self) -> dict: ... -``` - -**Dependencies**: None - ---- - -### 2.3 SSL/TLS Certificate Extraction - -**Goal**: Extract and display certificate details from TLS handshakes. - -**Scope**: -- Extract X.509 certificates from PCAP -- Parse certificate fields (subject, issuer, validity, SANs) -- Certificate chain validation -- Export certificates as PEM - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| Cert extractor | `app/pipeline/tls_certs.py` | New module | -| PyShark integration | `app/pipeline/pyshark_pass.py` | Add TLS parsing | -| UI display | `app/ui/tls_tab.py` | New tab or section | - -**Technical Details**: -```python -# app/pipeline/tls_certs.py -@dataclass -class Certificate: - subject: dict - issuer: dict - not_before: datetime - not_after: datetime - serial: str - sans: list[str] - fingerprint_sha256: str - -def extract_certificates(pcap_path: str) -> list[Certificate]: ... -def validate_chain(certs: list[Certificate]) -> dict: ... -``` - -**Dependencies**: `cryptography` (for X.509 parsing) - ---- - -## Phase 3: High Effort Features - -### 3.1 YARA Rule Scanning - -**Goal**: Scan carved files against YARA rules for malware detection. - -**Scope**: -- Integrate YARA engine -- Include default rule sets (malware signatures) -- Support custom rule upload -- Display matches with context - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| YARA scanner | `app/pipeline/yara_scan.py` | New module | -| Rule manager | `app/utils/yara_rules.py` | Load/manage rules | -| Default rules | `app/data/yara/` | Curated rule sets | -| UI integration | `app/ui/carve_tab.py` | Scan results display | - -**Technical Details**: -```python -# app/pipeline/yara_scan.py -class YARAScanner: - def __init__(self, rules_dir: str = None): ... - def add_rules(self, rules_path: str) -> None: ... - def scan_file(self, file_path: str) -> list[YARAMatch]: ... - def scan_directory(self, dir_path: str, phase: PhaseHandle) -> dict: ... - -@dataclass -class YARAMatch: - rule: str - tags: list[str] - strings: list[tuple[int, str, bytes]] - meta: dict -``` - -**Dependencies**: `yara-python` - ---- - -### 3.2 PDF Report Generation - -**Goal**: Generate professional PDF reports for documentation and sharing. - -**Scope**: -- Executive summary -- Detailed findings with visualizations -- IOC list -- Timeline of events -- Customizable branding - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| Report generator | `app/reports/pdf_report.py` | New module | -| Templates | `app/reports/templates/` | Jinja2 templates | -| Chart export | `app/utils/chart_export.py` | Export Plotly as images | -| UI button | `app/ui/report_tab.py` | Generate PDF button | - -**Technical Details**: -```python -# app/reports/pdf_report.py -class PDFReportGenerator: - def __init__(self, analysis_data: dict, template: str = "default"): ... - def generate(self, output_path: str) -> None: ... - - def _render_executive_summary(self) -> str: ... - def _render_flow_analysis(self) -> str: ... - def _render_osint_findings(self) -> str: ... - def _render_timeline(self) -> str: ... - def _render_ioc_table(self) -> str: ... -``` - -**Dependencies**: `weasyprint` or `reportlab`, `jinja2` - ---- - -### 3.3 Case Management - -**Goal**: Organize analyses into cases with notes, tags, and history. - -**Scope**: -- Create/manage cases -- Link multiple PCAPs to a case -- Add analyst notes and tags -- Search across cases -- Export case archive - -**Implementation**: - -| Component | File | Changes | -|-----------|------|---------| -| Case model | `app/models/case.py` | New module | -| Database | `app/db/cases.py` | SQLite persistence | -| Case API | `app/api/cases.py` | CRUD operations | -| UI | `app/ui/cases_tab.py` | Full case management UI | - -**Technical Details**: -```python -# app/models/case.py -@dataclass -class Case: - id: str - title: str - description: str - created_at: datetime - updated_at: datetime - tags: list[str] - pcaps: list[str] - notes: list[Note] - iocs: list[IOC] - status: CaseStatus - -# app/db/cases.py -class CaseDB: - def create(self, case: Case) -> str: ... - def get(self, case_id: str) -> Case | None: ... - def update(self, case: Case) -> None: ... - def delete(self, case_id: str) -> None: ... - def search(self, query: str, tags: list[str] = None) -> list[Case]: ... -``` - -**Dependencies**: `sqlite3`, possibly `sqlalchemy` - ---- - -## Implementation Priority Matrix - -| Feature | Effort | Value | Priority | -|---------|--------|-------|----------| -| CSV/JSON Export | Low | High | P0 | -| Config Persistence | Low | High | P0 | -| OSINT Caching | Low | Medium | P1 | -| JA3 Lookup | Low | Medium | P1 | -| DNS Carving | Medium | High | P1 | -| Multi-PCAP | Medium | High | P2 | -| TLS Cert Extraction | Medium | Medium | P2 | -| YARA Scanning | High | High | P2 | -| PDF Reports | High | Medium | P3 | -| Case Management | High | High | P3 | - ---- - -## Dependencies Summary - -**New packages required**: -``` -cryptography>=41.0.0 # Config encryption, TLS cert parsing -yara-python>=4.3.0 # YARA scanning (optional) -weasyprint>=60.0 # PDF generation (optional) -jinja2>=3.1.0 # Report templates (optional) -``` - ---- - -## Architecture Considerations - -### Module Organization - -``` -app/ -├── pipeline/ -│ ├── dns_carve.py # New: DNS carving -│ ├── dns_analysis.py # New: DGA/tunneling detection -│ ├── ja3.py # New: JA3 fingerprinting -│ ├── osint_cache.py # New: OSINT caching -│ ├── tls_certs.py # New: Certificate extraction -│ ├── yara_scan.py # New: YARA scanning -│ └── batch.py # New: Multi-PCAP processing -├── utils/ -│ ├── export.py # New: CSV/JSON export -│ ├── config_manager.py # New: Config persistence -│ └── yara_rules.py # New: YARA rule management -├── reports/ -│ ├── pdf_report.py # New: PDF generation -│ └── templates/ # New: Report templates -├── models/ -│ └── case.py # New: Case data model -├── db/ -│ └── cases.py # New: Case database -└── security/ - └── crypto.py # New: Encryption utilities -``` - -### Data Flow - -``` - ┌─────────────────┐ - │ PCAP Upload │ - └────────┬────────┘ - │ - ┌────────────────┼────────────────┐ - ▼ ▼ ▼ - ┌───────────┐ ┌───────────┐ ┌───────────┐ - │ Zeek │ │ PyShark │ │ Tshark │ - └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ - │ │ │ - │ ┌───────────┴───────────┐ │ - │ ▼ ▼ │ - │ ┌───────┐ ┌───────┐ │ - │ │ JA3 │ │ DNS │ │ - │ └───────┘ │ Carve │ │ - │ └───────┘ │ - │ │ - └──────────────┬──────────────────┘ - ▼ - ┌─────────────────────┐ - │ OSINT Enrichment │◄──── Cache - └──────────┬──────────┘ - │ - ┌────────────┼────────────┐ - ▼ ▼ ▼ - ┌──────────┐ ┌──────────┐ ┌──────────┐ - │ Beaconing│ │ YARA │ │ TLS │ - │ Detection│ │ Scanning│ │ Certs │ - └──────────┘ └──────────┘ └──────────┘ - │ - ▼ - ┌─────────────────────┐ - │ LLM Analysis │ - └──────────┬──────────┘ - │ - ┌────────────┼────────────┐ - ▼ ▼ ▼ - ┌──────────┐ ┌──────────┐ ┌──────────┐ - │ Export │ │ PDF │ │ Case │ - │ CSV/JSON │ │ Report │ │ Mgmt │ - └──────────┘ └──────────┘ └──────────┘ -``` - ---- - -## Version Milestones - -### v0.3.0 - Export & Persistence -- CSV/JSON Export -- Configuration Persistence -- OSINT Caching - -### v0.4.0 - Enhanced Analysis -- JA3/JA3S Fingerprinting -- DNS Carving & Analysis -- SSL/TLS Certificate Extraction - -### v0.5.0 - Advanced Features -- Multi-PCAP Batch Analysis -- YARA Rule Scanning - -### v1.0.0 - Enterprise Ready -- PDF Report Generation -- Case Management System -- Full documentation diff --git a/docs/PHASE4-IMPLEMENTATION-PLAN.md b/docs/PHASE4-IMPLEMENTATION-PLAN.md deleted file mode 100644 index d6e6d97..0000000 --- a/docs/PHASE4-IMPLEMENTATION-PLAN.md +++ /dev/null @@ -1,736 +0,0 @@ -# Phase 4 Implementation Plan - -## Overview - -Phase 4 focuses on **AI-enhanced analysis** and **export capabilities**, building upon the existing LLM integration and export utilities from Phase 1-3. - -### Goals -1. **AI Enhancement** - Leverage LLM for deeper analysis -2. **Export Enhancement** - Standard formats for threat intel sharing - ---- - -## 1. AI Enhancement - -### 1.1 MITRE ATT&CK Auto-Mapping - -**Purpose**: Automatically map detected behaviors to MITRE ATT&CK techniques. - -#### Detection Rules - -```python -# app/threat_intel/attack_mapping.py - -DETECTION_TO_ATTACK = { - # C2 Communication - "beacon_score": { - "threshold": 0.7, - "techniques": [ - {"id": "T1071.001", "name": "Web Protocols", "tactic": "command-and-control"}, - {"id": "T1571", "name": "Non-Standard Port", "tactic": "command-and-control"}, - ] - }, - # DNS-based - "dns_tunneling": { - "threshold": 0.6, - "techniques": [ - {"id": "T1071.004", "name": "DNS", "tactic": "command-and-control"}, - {"id": "T1048.003", "name": "Exfiltration Over DNS", "tactic": "exfiltration"}, - ] - }, - "dga_detected": { - "threshold": 0.7, - "techniques": [ - {"id": "T1568.002", "name": "Domain Generation Algorithms", "tactic": "command-and-control"}, - ] - }, - # TLS/Encryption - "self_signed_cert": { - "techniques": [ - {"id": "T1587.003", "name": "Digital Certificates", "tactic": "resource-development"}, - {"id": "T1573.002", "name": "Asymmetric Cryptography", "tactic": "command-and-control"}, - ] - }, - "ja3_malware": { - "techniques": [ - {"id": "T1071.001", "name": "Web Protocols", "tactic": "command-and-control"}, - ] - }, - # YARA matches - "yara_critical": { - "techniques": [ - {"id": "T1059", "name": "Command and Scripting Interpreter", "tactic": "execution"}, - {"id": "T1027", "name": "Obfuscated Files or Information", "tactic": "defense-evasion"}, - ] - }, -} -``` - -#### Core Functions - -```python -@dataclass -class TechniqueMatch: - technique_id: str # T1071.001 - technique_name: str # Web Protocols - tactic: str # command-and-control - confidence: float # 0.0 - 1.0 - evidence: list[str] # What triggered this detection - -@dataclass -class AttackMapping: - techniques: list[TechniqueMatch] - tactics_summary: dict[str, int] # tactic -> count - kill_chain_phase: str # reconnaissance, initial-access, etc. - overall_severity: str # low, medium, high, critical - -class ATTACKMapper: - def __init__(self): - """Initialize with detection rules.""" - - def map_analysis(self, - features: dict, - dns_analysis: dict, - tls_analysis: dict, - yara_results: dict, - beacon_results: list) -> AttackMapping: - """Map analysis results to ATT&CK techniques.""" - - def enhance_with_llm(self, - mapping: AttackMapping, - report_md: str) -> AttackMapping: - """Use LLM to refine mapping and add context.""" -``` - -#### LLM Enhancement Prompt - -```python -ATTACK_MAPPING_PROMPT = """ -Based on the analysis findings below, identify additional MITRE ATT&CK techniques -that may apply. For each technique, provide: -1. Technique ID (e.g., T1071.001) -2. Confidence level (0.0-1.0) -3. Evidence from the analysis - -Current findings: -{analysis_summary} - -Already identified techniques: -{current_techniques} - -Respond in JSON format: -{ - "additional_techniques": [ - {"id": "T1xxx", "confidence": 0.8, "evidence": "..."} - ], - "attack_narrative": "Brief description of the likely attack chain..." -} -""" -``` - ---- - -### 1.2 Attack Timeline Narrative - -**Purpose**: Generate a human-readable story of the attack progression. - -#### Core Functions - -```python -class AttackNarrator: - def __init__(self, llm_client): - """Initialize with LLM client.""" - - def generate_narrative(self, - features: dict, - dns_analysis: dict, - tls_analysis: dict, - attack_mapping: AttackMapping, - timeline_events: list) -> str: - """Generate attack story using LLM.""" - - def create_timeline(self, features: dict) -> list[TimelineEvent]: - """Extract chronological events from analysis.""" - -@dataclass -class TimelineEvent: - timestamp: datetime - event_type: str # connection, dns_query, file_download, alert - description: str - severity: str - source_ip: str - dest_ip: str - iocs: list[str] -``` - -#### LLM Narrative Prompt - -```python -NARRATIVE_PROMPT = """ -Based on the timeline of events and detected techniques, write a concise -attack narrative that explains: - -1. How the attack likely began (initial access) -2. What the attacker did (execution, persistence) -3. How they communicated with C2 (command and control) -4. What data may have been exfiltrated (if applicable) -5. Current status and recommended actions - -Timeline: -{timeline_events} - -Detected ATT&CK Techniques: -{techniques} - -Key IOCs: -{iocs} - -Write in {language}, using professional security terminology. -Keep the narrative to 3-5 paragraphs. -""" -``` - ---- - -### 1.3 IOC Priority Scoring - -**Purpose**: Score IOCs by importance to help analysts focus on what matters. - -#### Scoring Factors - -```python -IOC_SCORING_WEIGHTS = { - # OSINT signals - "vt_detections": 0.25, # VirusTotal detection ratio - "greynoise_malicious": 0.20, # GreyNoise classification - "abuseipdb_score": 0.15, # AbuseIPDB confidence - - # Behavioral signals - "beacon_score": 0.15, # C2 beaconing likelihood - "connection_count": 0.10, # Frequency of communication - "data_volume": 0.05, # Amount of data transferred - - # Context signals - "ja3_malware_match": 0.10, # Known malicious fingerprint - "dga_match": 0.05, # DGA domain - "self_signed_cert": 0.05, # Suspicious certificate -} - -class IOCScorer: - def __init__(self): - """Initialize scorer.""" - - def score_ioc(self, - ioc_value: str, - ioc_type: str, - osint_data: dict, - behavioral_data: dict) -> float: - """Calculate priority score 0.0-1.0.""" - - def rank_iocs(self, - iocs: list[dict], - osint: dict, - features: dict) -> list[dict]: - """Return IOCs sorted by priority with scores.""" - - def explain_score(self, ioc: dict) -> str: - """Generate human-readable explanation of score.""" -``` - -#### Output Example - -```python -{ - "ioc": "185.220.101.45", - "type": "ip", - "priority_score": 0.92, - "priority_label": "critical", - "factors": { - "vt_detections": {"value": 45, "contribution": 0.25}, - "greynoise": {"value": "malicious", "contribution": 0.20}, - "beacon_score": {"value": 0.85, "contribution": 0.15}, - "ja3_match": {"value": "Cobalt Strike", "contribution": 0.10} - }, - "recommendation": "Immediate block recommended" -} -``` - ---- - -### 1.4 Interactive Q&A - -**Purpose**: Allow analysts to ask questions about the analysis results. - -#### Architecture - -```python -class AnalysisQA: - def __init__(self, llm_client, analysis_context: dict): - """Initialize with analysis data as context.""" - self.context = analysis_context - self.conversation_history = [] - - def ask(self, question: str) -> str: - """Ask a question about the analysis.""" - - def get_suggested_questions(self) -> list[str]: - """Return relevant questions based on findings.""" - -# Suggested questions based on findings -SUGGESTED_QUESTIONS = { - "beacon_detected": [ - "What is the beaconing interval pattern?", - "Which internal hosts are beaconing?", - "What C2 infrastructure is being used?", - ], - "yara_match": [ - "What malware was detected?", - "Which files triggered the YARA rules?", - "What are the capabilities of this malware?", - ], - "dga_detected": [ - "How many DGA domains were found?", - "What DGA algorithm might be in use?", - "Are any DGA domains resolving?", - ], -} -``` - -#### UI Integration - -```python -def render_qa_section(): - """Render Q&A interface in Streamlit.""" - st.markdown("### Ask About This Analysis") - - # Show suggested questions - suggestions = qa.get_suggested_questions() - if suggestions: - st.caption("Suggested questions:") - for q in suggestions: - if st.button(q, key=f"q_{hash(q)}"): - st.session_state["qa_question"] = q - - # Question input - question = st.text_input( - "Your question:", - value=st.session_state.get("qa_question", ""), - placeholder="e.g., What is the most likely attack vector?" - ) - - if st.button("Ask"): - with st.spinner("Analyzing..."): - answer = qa.ask(question) - st.markdown(answer) -``` - ---- - -## 2. Export Enhancement - -### 2.1 IOC List Export - -**Purpose**: Export IOCs in multiple formats for easy sharing. - -#### Supported Formats - -| Format | Use Case | -|--------|----------| -| CSV | Spreadsheets, quick review | -| JSON | Programmatic processing | -| TXT | Firewall block lists | -| STIX 2.1 | Threat intel platforms | - -#### Core Functions - -```python -# app/utils/ioc_export.py - -class IOCExporter: - def __init__(self, features: dict, osint: dict, scores: dict = None): - """Initialize with analysis data.""" - - def extract_iocs(self) -> list[dict]: - """Extract all IOCs from analysis.""" - - def export_csv(self, ioc_types: list[str] = None) -> bytes: - """Export to CSV format.""" - - def export_json(self, ioc_types: list[str] = None) -> bytes: - """Export to JSON format.""" - - def export_txt(self, ioc_types: list[str] = None) -> bytes: - """Export plain text list (one IOC per line).""" - - def export_stix(self, ioc_types: list[str] = None) -> bytes: - """Export as STIX 2.1 Bundle.""" - -# IOC structure -@dataclass -class IOCRecord: - type: str # ip, domain, hash, ja3, url - value: str - context: str # Where it was found - first_seen: datetime - last_seen: datetime - priority_score: float - osint_summary: dict - tags: list[str] -``` - -#### UI Integration - -```python -def render_ioc_export(): - """Render IOC export controls.""" - st.markdown("### Export IOCs") - - col1, col2 = st.columns(2) - - with col1: - ioc_types = st.multiselect( - "IOC Types", - ["ip", "domain", "hash", "ja3", "url"], - default=["ip", "domain"] - ) - - with col2: - format_choice = st.selectbox( - "Format", - ["CSV", "JSON", "Plain Text", "STIX 2.1"] - ) - - min_score = st.slider( - "Minimum Priority Score", - 0.0, 1.0, 0.0, - help="Filter to high-priority IOCs only" - ) - - if st.button("Export"): - exporter = IOCExporter(features, osint, scores) - - if format_choice == "CSV": - data = exporter.export_csv(ioc_types) - filename = "iocs.csv" - mime = "text/csv" - elif format_choice == "JSON": - data = exporter.export_json(ioc_types) - filename = "iocs.json" - mime = "application/json" - elif format_choice == "Plain Text": - data = exporter.export_txt(ioc_types) - filename = "iocs.txt" - mime = "text/plain" - else: # STIX - data = exporter.export_stix(ioc_types) - filename = "iocs_stix.json" - mime = "application/json" - - st.download_button( - f"Download {filename}", - data=data, - file_name=filename, - mime=mime - ) -``` - ---- - -### 2.2 ATT&CK Navigator Layer - -**Purpose**: Export detected techniques as ATT&CK Navigator layer for visualization. - -#### Navigator Layer Format - -```python -def export_navigator_layer(mapping: AttackMapping, - name: str = "PCAP Analysis") -> dict: - """ - Generate ATT&CK Navigator layer JSON. - Can be imported directly into https://mitre-attack.github.io/attack-navigator/ - """ - techniques = [] - - for tech in mapping.techniques: - techniques.append({ - "techniqueID": tech.technique_id, - "tactic": tech.tactic, - "score": int(tech.confidence * 100), - "color": _severity_to_color(tech.confidence), - "comment": "; ".join(tech.evidence), - "enabled": True, - }) - - return { - "name": name, - "versions": { - "attack": "14", - "navigator": "4.9.1", - "layer": "4.5" - }, - "domain": "enterprise-attack", - "description": f"Generated by PCAP Hunter on {datetime.now().isoformat()}", - "techniques": techniques, - "gradient": { - "colors": ["#ffffff", "#ffeb3b", "#ff9800", "#f44336"], - "minValue": 0, - "maxValue": 100 - }, - "legendItems": [ - {"label": "Low Confidence", "color": "#ffeb3b"}, - {"label": "Medium Confidence", "color": "#ff9800"}, - {"label": "High Confidence", "color": "#f44336"}, - ] - } -``` - -#### UI Integration - -```python -def render_attack_export(): - """Render ATT&CK export controls.""" - st.markdown("### Export ATT&CK Mapping") - - layer_name = st.text_input( - "Layer Name", - value=f"PCAP Analysis - {datetime.now().strftime('%Y-%m-%d')}" - ) - - if st.button("Export Navigator Layer"): - layer = export_navigator_layer(attack_mapping, layer_name) - - st.download_button( - "Download Navigator Layer", - data=json.dumps(layer, indent=2), - file_name="attack_layer.json", - mime="application/json" - ) - - st.info("Import this file at: https://mitre-attack.github.io/attack-navigator/") -``` - ---- - -### 2.3 STIX 2.1 Export - -**Purpose**: Export findings in STIX 2.1 format for threat intel platforms. - -#### STIX Objects Generated - -| STIX Type | Source | -|-----------|--------| -| Indicator | IOCs (IP, Domain, Hash) | -| Malware | YARA matches | -| Attack Pattern | ATT&CK techniques | -| Relationship | Links between objects | -| Report | Analysis summary | - -#### Core Functions - -```python -# app/utils/stix_export.py - -from stix2 import ( - Bundle, Indicator, Malware, AttackPattern, - Relationship, Report, Identity -) - -class STIXExporter: - def __init__(self, - features: dict, - osint: dict, - attack_mapping: AttackMapping, - yara_results: dict): - """Initialize with analysis data.""" - self.identity = self._create_identity() - - def _create_identity(self) -> Identity: - """Create identity for the analysis tool.""" - return Identity( - name="PCAP Hunter", - identity_class="tool" - ) - - def create_indicators(self) -> list[Indicator]: - """Create STIX Indicators from IOCs.""" - - def create_malware(self) -> list[Malware]: - """Create STIX Malware from YARA matches.""" - - def create_attack_patterns(self) -> list[AttackPattern]: - """Create STIX Attack Patterns from ATT&CK mapping.""" - - def create_relationships(self) -> list[Relationship]: - """Create relationships between objects.""" - - def create_report(self, title: str) -> Report: - """Create summary report object.""" - - def export_bundle(self) -> Bundle: - """Export complete STIX Bundle.""" -``` - -#### STIX Indicator Example - -```python -def _ioc_to_indicator(self, ioc: dict) -> Indicator: - """Convert IOC to STIX Indicator.""" - - # Build pattern based on IOC type - if ioc["type"] == "ip": - pattern = f"[ipv4-addr:value = '{ioc['value']}']" - elif ioc["type"] == "domain": - pattern = f"[domain-name:value = '{ioc['value']}']" - elif ioc["type"] == "hash": - pattern = f"[file:hashes.'SHA-256' = '{ioc['value']}']" - - return Indicator( - name=f"Malicious {ioc['type']}: {ioc['value']}", - pattern=pattern, - pattern_type="stix", - valid_from=datetime.now(), - labels=ioc.get("tags", []), - confidence=int(ioc.get("priority_score", 0.5) * 100), - created_by_ref=self.identity.id, - external_references=[ - {"source_name": "PCAP Hunter", "description": ioc.get("context", "")} - ] - ) -``` - ---- - -## 3. Implementation Order - -### Step 1: MITRE ATT&CK Mapping (Day 1-2) -1. Create `app/threat_intel/attack_mapping.py` -2. Define detection-to-technique rules -3. Integrate with LLM for enhancement -4. Add ATT&CK section to PDF report -5. Write tests - -### Step 2: IOC Export (Day 2-3) -1. Create `app/utils/ioc_export.py` -2. Implement CSV/JSON/TXT export -3. Add export UI to Dashboard -4. Write tests - -### Step 3: IOC Priority Scoring (Day 3-4) -1. Create `app/analysis/ioc_scorer.py` -2. Implement scoring algorithm -3. Integrate with export -4. Add score display to UI -5. Write tests - -### Step 4: ATT&CK Navigator Export (Day 4) -1. Implement Navigator layer format -2. Add export button to UI -3. Write tests - -### Step 5: Attack Narrative (Day 5) -1. Enhance LLM prompts -2. Add narrative section to report -3. Write tests - -### Step 6: Interactive Q&A (Day 6) -1. Create `app/llm/qa.py` -2. Implement conversation context -3. Add Q&A UI section -4. Write tests - -### Step 7: STIX Export (Day 7-8) -1. Create `app/utils/stix_export.py` -2. Implement STIX object creation -3. Add to export options -4. Write tests - ---- - -## 4. New Files - -``` -app/ -├── threat_intel/ -│ ├── __init__.py -│ └── attack_mapping.py # ATT&CK mapping engine -├── analysis/ -│ ├── __init__.py -│ ├── ioc_scorer.py # IOC priority scoring -│ └── narrator.py # Attack narrative generation -├── llm/ -│ └── qa.py # Interactive Q&A -└── utils/ - ├── ioc_export.py # IOC export (CSV/JSON/TXT) - ├── stix_export.py # STIX 2.1 export - └── navigator_export.py # ATT&CK Navigator layer - -tests/ -├── test_attack_mapping.py -├── test_ioc_scorer.py -├── test_ioc_export.py -├── test_stix_export.py -└── test_qa.py -``` - ---- - -## 5. Dependencies - -```toml -# pyproject.toml additions -[project.optional-dependencies] -phase4 = [ - "stix2>=3.0.0", # STIX 2.1 export -] -``` - ---- - -## 6. UI Changes - -### Dashboard Tab Additions - -```python -# New sections in layout.py - -# After OSINT section: -with st.expander("🎯 ATT&CK Mapping", expanded=True): - render_attack_mapping(attack_mapping) - render_attack_export() - -# After Report section: -with st.expander("📤 Export IOCs", expanded=False): - render_ioc_export() - -# New tab or section: -with st.expander("💬 Ask About Analysis", expanded=False): - render_qa_section() -``` - -### PDF Report Additions - -``` -New sections: -6. MITRE ATT&CK Mapping - - Detected techniques table - - Tactics coverage - - Attack narrative - -7. IOC Priority Summary - - Top 10 critical IOCs - - Score explanations -``` - ---- - -## 7. Success Metrics - -| Feature | Success Criteria | -|---------|------------------| -| ATT&CK Mapping | Correctly identifies 80%+ of applicable techniques | -| IOC Export | All formats export without errors | -| Priority Scoring | High-priority IOCs have OSINT correlation | -| Navigator Export | Layer imports correctly in Navigator | -| Attack Narrative | Coherent story matching timeline | -| Interactive Q&A | Accurate answers to common questions | -| STIX Export | Valid STIX 2.1 Bundle | diff --git a/docs/TEST-PLAN.md b/docs/TEST-PLAN.md deleted file mode 100644 index a006cd0..0000000 --- a/docs/TEST-PLAN.md +++ /dev/null @@ -1,752 +0,0 @@ -# PCAP Hunter Test Plan - -## Overview - -This document outlines the testing strategy for new features in PCAP Hunter. It covers unit tests, integration tests, and end-to-end testing approaches. - ---- - -## Testing Philosophy - -1. **Test-First Development**: Write tests before implementation where possible -2. **High Coverage on Critical Paths**: Security and data processing code must have >90% coverage -3. **Mocked External Dependencies**: API calls, file system operations use mocks -4. **Realistic Test Data**: Use sanitized PCAP samples for integration tests - ---- - -## Test Infrastructure - -### Directory Structure - -``` -tests/ -├── conftest.py # Shared fixtures -├── fixtures/ # Test data -│ ├── sample.pcap # Small test PCAP -│ ├── dns_tunnel.pcap # DNS tunneling sample -│ ├── beacon.pcap # C2 beaconing sample -│ └── malware_carved.bin # Test carved file -├── unit/ # Unit tests -│ ├── test_export.py -│ ├── test_config_manager.py -│ ├── test_osint_cache.py -│ ├── test_ja3.py -│ ├── test_dns_analysis.py -│ ├── test_tls_certs.py -│ ├── test_yara_scan.py -│ └── test_case_model.py -├── integration/ # Integration tests -│ ├── test_pipeline_export.py -│ ├── test_batch_analysis.py -│ └── test_case_workflow.py -└── e2e/ # End-to-end tests - └── test_full_analysis.py -``` - -### Shared Fixtures (`conftest.py`) - -```python -import pytest -from pathlib import Path - -@pytest.fixture -def sample_pcap(): - """Path to small test PCAP file.""" - return Path(__file__).parent / "fixtures" / "sample.pcap" - -@pytest.fixture -def sample_flows(): - """Sample flow data for testing.""" - return [ - {"src": "192.168.1.100", "dst": "8.8.8.8", "sport": 54321, "dport": 53, - "proto": "DNS", "pkt_times": [1.0, 2.0, 3.0]}, - {"src": "192.168.1.100", "dst": "1.1.1.1", "sport": 54322, "dport": 443, - "proto": "TLS", "pkt_times": [1.5, 2.5, 3.5]}, - ] - -@pytest.fixture -def mock_osint_response(): - """Sample OSINT API response.""" - return { - "greynoise": {"seen": True, "classification": "malicious"}, - "abuseipdb": {"data": {"abuseConfidenceScore": 75}}, - } - -@pytest.fixture -def temp_data_dir(tmp_path): - """Temporary directory for test outputs.""" - data_dir = tmp_path / "data" - data_dir.mkdir() - return data_dir -``` - ---- - -## Phase 1: Quick Wins Test Cases - -### 1.1 CSV/JSON Export Tests - -**File**: `tests/unit/test_export.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| EXP-001 | test_export_flows_csv | Export flows to CSV | List of flow dicts | Valid CSV bytes with headers | -| EXP-002 | test_export_flows_json | Export flows to JSON | List of flow dicts | Valid JSON bytes | -| EXP-003 | test_export_empty_data | Handle empty input | Empty list | Empty CSV/JSON structure | -| EXP-004 | test_export_special_chars | Handle special characters | Flows with unicode | Properly escaped output | -| EXP-005 | test_export_nested_data | Handle nested dicts | OSINT results | Flattened CSV / nested JSON | -| EXP-006 | test_export_large_dataset | Performance test | 100k flows | Completes in <5s | - -```python -# tests/unit/test_export.py -import json -import csv -from io import StringIO -from app.utils.export import export_to_csv, export_to_json - -def test_export_flows_csv(sample_flows): - result = export_to_csv(sample_flows, "flows.csv") - assert isinstance(result, bytes) - reader = csv.DictReader(StringIO(result.decode())) - rows = list(reader) - assert len(rows) == 2 - assert rows[0]["src"] == "192.168.1.100" - -def test_export_flows_json(sample_flows): - result = export_to_json(sample_flows, "flows.json") - data = json.loads(result) - assert len(data) == 2 - assert data[0]["proto"] == "DNS" - -def test_export_empty_data(): - result = export_to_csv([], "empty.csv") - assert result == b"" # or just headers - -def test_export_special_chars(): - flows = [{"src": "192.168.1.1", "note": "Test \u2605 unicode"}] - result = export_to_csv(flows, "special.csv") - assert "unicode" in result.decode("utf-8") -``` - ---- - -### 1.2 Configuration Persistence Tests - -**File**: `tests/unit/test_config_manager.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| CFG-001 | test_save_load_config | Round-trip save/load | Config dict | Same values | -| CFG-002 | test_encrypt_api_keys | API keys encrypted | Config with keys | Encrypted in file | -| CFG-003 | test_missing_config_file | Handle missing file | No file exists | Returns defaults | -| CFG-004 | test_corrupted_config | Handle corrupted JSON | Invalid JSON | Returns defaults, logs error | -| CFG-005 | test_partial_config | Merge with defaults | Partial config | Merged with defaults | -| CFG-006 | test_config_isolation | Per-project configs | Two projects | Separate configs | - -```python -# tests/unit/test_config_manager.py -from app.utils.config_manager import ConfigManager - -def test_save_load_config(temp_data_dir): - config_path = temp_data_dir / ".config.json" - manager = ConfigManager(config_path) - - original = {"llm_endpoint": "http://localhost:1234", "vt_key": "secret123"} - manager.save(original) - - loaded = manager.load() - assert loaded["llm_endpoint"] == original["llm_endpoint"] - assert loaded["vt_key"] == original["vt_key"] - -def test_encrypt_api_keys(temp_data_dir): - config_path = temp_data_dir / ".config.json" - manager = ConfigManager(config_path) - - manager.save({"vt_key": "my_secret_key"}) - - # Read raw file - API key should be encrypted - raw = config_path.read_text() - assert "my_secret_key" not in raw - assert "encrypted:" in raw or "ENC[" in raw - -def test_missing_config_file(temp_data_dir): - config_path = temp_data_dir / "nonexistent.json" - manager = ConfigManager(config_path) - - config = manager.load() - assert config == manager.defaults -``` - ---- - -### 1.3 OSINT Cache Tests - -**File**: `tests/unit/test_osint_cache.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| OSC-001 | test_cache_miss | Query uncached IP | New IP | None | -| OSC-002 | test_cache_hit | Query cached IP | Cached IP | Cached data | -| OSC-003 | test_cache_expiry | TTL expiration | Expired entry | None | -| OSC-004 | test_cache_invalidate | Manual invalidation | Cached IP | Entry removed | -| OSC-005 | test_cache_stats | Get cache statistics | After operations | Correct counts | -| OSC-006 | test_concurrent_access | Thread safety | Parallel writes | No corruption | - -```python -# tests/unit/test_osint_cache.py -import time -from app.pipeline.osint_cache import OSINTCache - -def test_cache_miss(temp_data_dir): - cache = OSINTCache(temp_data_dir / "osint.db") - result = cache.get("8.8.8.8", "greynoise") - assert result is None - -def test_cache_hit(temp_data_dir, mock_osint_response): - cache = OSINTCache(temp_data_dir / "osint.db") - cache.set("8.8.8.8", "greynoise", mock_osint_response["greynoise"]) - - result = cache.get("8.8.8.8", "greynoise") - assert result == mock_osint_response["greynoise"] - -def test_cache_expiry(temp_data_dir): - cache = OSINTCache(temp_data_dir / "osint.db", ttl_seconds=1) - cache.set("8.8.8.8", "greynoise", {"test": True}) - - time.sleep(1.5) - result = cache.get("8.8.8.8", "greynoise") - assert result is None - -def test_cache_invalidate(temp_data_dir): - cache = OSINTCache(temp_data_dir / "osint.db") - cache.set("8.8.8.8", "greynoise", {"test": True}) - cache.set("1.1.1.1", "greynoise", {"test": True}) - - count = cache.invalidate("8.8.8.8") - assert count == 1 - assert cache.get("8.8.8.8", "greynoise") is None - assert cache.get("1.1.1.1", "greynoise") is not None -``` - ---- - -### 1.4 JA3 Fingerprint Tests - -**File**: `tests/unit/test_ja3.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| JA3-001 | test_calculate_ja3 | Calculate JA3 hash | TLS params | Correct MD5 hash | -| JA3-002 | test_known_fingerprint | Lookup known JA3 | Chrome JA3 | "Google Chrome" | -| JA3-003 | test_unknown_fingerprint | Lookup unknown JA3 | Random hash | None | -| JA3-004 | test_ja3s_calculation | Calculate JA3S | Server params | Correct hash | -| JA3-005 | test_malware_fingerprint | Detect malware JA3 | Cobalt Strike JA3 | Malware match | - -```python -# tests/unit/test_ja3.py -from app.pipeline.ja3 import calculate_ja3, lookup_ja3 - -def test_calculate_ja3(): - # Known Chrome JA3 parameters - ja3_hash = calculate_ja3( - version="771", # TLS 1.2 - ciphers=["49195", "49196", "49199", "49200"], - extensions=["0", "23", "65281"], - curves=["29", "23", "24"], - point_formats=["0"] - ) - assert len(ja3_hash) == 32 # MD5 hash - assert ja3_hash.isalnum() - -def test_known_fingerprint(): - # Well-known Chrome JA3 - chrome_ja3 = "769,47-53-5-10-49171-49172-49161-49162,0-10-11,23-24,0" - result = lookup_ja3(calculate_ja3_from_string(chrome_ja3)) - assert result is not None - assert "Chrome" in result.get("client", "") - -def test_unknown_fingerprint(): - result = lookup_ja3("0" * 32) - assert result is None - -def test_malware_fingerprint(): - # Known Cobalt Strike JA3 - cobalt_ja3 = "72a589da586844d7f0818ce684948eea" - result = lookup_ja3(cobalt_ja3) - assert result is not None - assert result.get("malware", False) or "Cobalt" in result.get("notes", "") -``` - ---- - -## Phase 2: Medium Effort Test Cases - -### 2.1 DNS Analysis Tests - -**File**: `tests/unit/test_dns_analysis.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| DNS-001 | test_normal_domain | Score normal domain | "google.com" | Low DGA score (<0.3) | -| DNS-002 | test_dga_domain | Detect DGA domain | "x7k9m2p4.com" | High DGA score (>0.7) | -| DNS-003 | test_subdomain_entropy | High entropy subdomain | Long random subdomain | Tunneling flag | -| DNS-004 | test_fast_flux | Detect fast-flux | Many IPs, short TTL | Fast-flux detected | -| DNS-005 | test_txt_tunneling | TXT record tunneling | Large TXT responses | Tunneling detected | - -```python -# tests/unit/test_dns_analysis.py -from app.pipeline.dns_analysis import detect_dga, detect_tunneling, detect_fast_flux - -def test_normal_domain(): - score = detect_dga("google.com") - assert score < 0.3 - -def test_dga_domain(): - # Typical DGA-generated domain - score = detect_dga("x7k9m2p4q1.com") - assert score > 0.7 - -def test_subdomain_entropy(): - dns_records = [ - {"query": "aGVsbG8gd29ybGQgdGhpcyBpcyBhIHRlc3Q.evil.com", "type": "A"} - ] - result = detect_tunneling(dns_records) - assert result["high_entropy_subdomains"] > 0 - -def test_fast_flux(): - # Same domain, many different IPs - responses = [ - {"domain": "malware.com", "ip": f"1.2.3.{i}", "ttl": 60} - for i in range(20) - ] - result = detect_fast_flux("malware.com", responses) - assert result is True - -def test_txt_tunneling(): - dns_records = [ - {"query": "data.evil.com", "type": "TXT", - "answer": "VGhpcyBpcyBhIHZlcnkgbG9uZyBiYXNlNjQgZW5jb2RlZCBzdHJpbmc="} - ] - result = detect_tunneling(dns_records) - assert result["suspicious_txt"] > 0 -``` - ---- - -### 2.2 TLS Certificate Tests - -**File**: `tests/unit/test_tls_certs.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| TLS-001 | test_extract_cert | Extract certificate | PCAP with TLS | Certificate object | -| TLS-002 | test_expired_cert | Detect expired cert | Expired cert | is_expired=True | -| TLS-003 | test_self_signed | Detect self-signed | Self-signed cert | is_self_signed=True | -| TLS-004 | test_cert_chain | Validate chain | Full chain | chain_valid=True | -| TLS-005 | test_san_extraction | Extract SANs | Cert with SANs | List of SANs | - -```python -# tests/unit/test_tls_certs.py -from datetime import datetime, timedelta -from app.pipeline.tls_certs import Certificate, validate_chain - -def test_extract_cert(sample_pcap): - from app.pipeline.tls_certs import extract_certificates - certs = extract_certificates(str(sample_pcap)) - assert len(certs) > 0 - assert isinstance(certs[0], Certificate) - -def test_expired_cert(): - cert = Certificate( - subject={"CN": "test.com"}, - issuer={"CN": "Test CA"}, - not_before=datetime.now() - timedelta(days=365), - not_after=datetime.now() - timedelta(days=1), - serial="1234", - sans=["test.com"], - fingerprint_sha256="abc123" - ) - assert cert.is_expired() - -def test_self_signed(): - cert = Certificate( - subject={"CN": "test.com"}, - issuer={"CN": "test.com"}, # Same as subject - not_before=datetime.now(), - not_after=datetime.now() + timedelta(days=365), - serial="1234", - sans=["test.com"], - fingerprint_sha256="abc123" - ) - assert cert.is_self_signed() -``` - ---- - -### 2.3 Batch Analysis Tests - -**File**: `tests/integration/test_batch_analysis.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| BAT-001 | test_multi_pcap | Process multiple PCAPs | 3 PCAPs | Combined results | -| BAT-002 | test_ip_correlation | Correlate IPs across | Related PCAPs | Correlation map | -| BAT-003 | test_timeline_merge | Merge timelines | Overlapping times | Unified timeline | -| BAT-004 | test_progress_tracking | Track batch progress | 5 PCAPs | 5 phase updates | - -```python -# tests/integration/test_batch_analysis.py -from app.pipeline.batch import BatchProcessor - -def test_multi_pcap(temp_data_dir): - pcaps = [temp_data_dir / f"test{i}.pcap" for i in range(3)] - # Create test PCAPs... - - processor = BatchProcessor([str(p) for p in pcaps]) - result = processor.process_all(phase=None) - - assert "combined_flows" in result - assert len(result["file_results"]) == 3 - -def test_ip_correlation(temp_data_dir): - # Two PCAPs with same malicious IP - processor = BatchProcessor(["pcap1.pcap", "pcap2.pcap"]) - result = processor.correlate() - - assert "shared_ips" in result - assert "shared_domains" in result -``` - ---- - -## Phase 3: High Effort Test Cases - -### 3.1 YARA Scanning Tests - -**File**: `tests/unit/test_yara_scan.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| YAR-001 | test_load_rules | Load YARA rules | Rule file | Rules compiled | -| YAR-002 | test_scan_clean | Scan clean file | Benign file | No matches | -| YAR-003 | test_scan_malware | Scan malware | Known malware | Matches found | -| YAR-004 | test_custom_rules | Use custom rules | User rule | Rule applied | -| YAR-005 | test_scan_directory | Batch scan | Directory | All files scanned | - -```python -# tests/unit/test_yara_scan.py -from app.pipeline.yara_scan import YARAScanner, YARAMatch - -def test_load_rules(temp_data_dir): - rule_file = temp_data_dir / "test.yar" - rule_file.write_text(''' -rule TestRule { - strings: - $a = "malicious_string" - condition: - $a -} -''') - scanner = YARAScanner() - scanner.add_rules(str(rule_file)) - assert scanner.rule_count > 0 - -def test_scan_clean(temp_data_dir): - clean_file = temp_data_dir / "clean.txt" - clean_file.write_text("This is a normal file with no malicious content.") - - scanner = YARAScanner() - matches = scanner.scan_file(str(clean_file)) - assert len(matches) == 0 - -def test_scan_malware(temp_data_dir): - malware_file = temp_data_dir / "malware.bin" - malware_file.write_text("Contains malicious_string for testing") - - rule_file = temp_data_dir / "test.yar" - rule_file.write_text(''' -rule TestMalware { - strings: - $a = "malicious_string" - condition: - $a -} -''') - - scanner = YARAScanner() - scanner.add_rules(str(rule_file)) - matches = scanner.scan_file(str(malware_file)) - - assert len(matches) == 1 - assert matches[0].rule == "TestMalware" -``` - ---- - -### 3.2 Case Management Tests - -**File**: `tests/unit/test_case_model.py` & `tests/integration/test_case_workflow.py` - -| Test ID | Test Name | Description | Input | Expected Output | -|---------|-----------|-------------|-------|-----------------| -| CAS-001 | test_create_case | Create new case | Case data | Case ID returned | -| CAS-002 | test_add_pcap | Add PCAP to case | Case + PCAP | PCAP linked | -| CAS-003 | test_add_note | Add analyst note | Case + note | Note saved | -| CAS-004 | test_search_cases | Search by keyword | Search query | Matching cases | -| CAS-005 | test_tag_filtering | Filter by tags | Tag list | Tagged cases | -| CAS-006 | test_export_case | Export case archive | Case ID | ZIP archive | - -```python -# tests/unit/test_case_model.py -from app.models.case import Case, Note, CaseStatus -from app.db.cases import CaseDB - -def test_create_case(temp_data_dir): - db = CaseDB(temp_data_dir / "cases.db") - - case = Case( - id=None, - title="Suspicious Traffic Investigation", - description="Investigating potential C2 traffic", - tags=["c2", "malware"], - pcaps=[], - notes=[], - status=CaseStatus.OPEN - ) - - case_id = db.create(case) - assert case_id is not None - - retrieved = db.get(case_id) - assert retrieved.title == "Suspicious Traffic Investigation" - -def test_add_note(temp_data_dir): - db = CaseDB(temp_data_dir / "cases.db") - case_id = db.create(Case(title="Test Case", ...)) - - note = Note( - author="analyst1", - content="Found suspicious beaconing pattern", - created_at=datetime.now() - ) - - db.add_note(case_id, note) - case = db.get(case_id) - assert len(case.notes) == 1 - assert "beaconing" in case.notes[0].content - -def test_search_cases(temp_data_dir): - db = CaseDB(temp_data_dir / "cases.db") - - db.create(Case(title="APT29 Investigation", tags=["apt", "russia"])) - db.create(Case(title="Ransomware Incident", tags=["ransomware"])) - db.create(Case(title="APT28 Analysis", tags=["apt", "russia"])) - - results = db.search("APT") - assert len(results) == 2 - - results = db.search(tags=["russia"]) - assert len(results) == 2 -``` - ---- - -## Integration Test Strategy - -### Pipeline Integration Tests - -**File**: `tests/integration/test_pipeline_export.py` - -```python -def test_full_analysis_to_export(sample_pcap, temp_data_dir): - """Test complete flow: PCAP -> Analysis -> Export""" - from app.pipeline import run_full_analysis - from app.utils.export import export_to_json - - # Run analysis - result = run_full_analysis(str(sample_pcap), str(temp_data_dir)) - - # Export results - json_data = export_to_json(result["flows"], "flows.json") - - # Verify export - import json - exported = json.loads(json_data) - assert len(exported) == len(result["flows"]) -``` - ---- - -## End-to-End Test Strategy - -### Full Workflow Test - -**File**: `tests/e2e/test_full_analysis.py` - -```python -def test_complete_workflow(sample_pcap, temp_data_dir): - """Simulate complete user workflow""" - # 1. Upload PCAP - # 2. Configure analysis - # 3. Run pipeline - # 4. Verify results - # 5. Export report - pass -``` - ---- - -## Test Data Requirements - -### Sample PCAPs Needed - -| Filename | Purpose | Size | Content | -|----------|---------|------|---------| -| `sample.pcap` | Basic tests | <1MB | HTTP, DNS, TLS traffic | -| `beacon.pcap` | C2 detection | <1MB | Regular interval connections | -| `dns_tunnel.pcap` | DNS analysis | <1MB | DNS tunneling traffic | -| `malware_pcap` | YARA testing | <1MB | Traffic with malware indicators | -| `multi_protocol.pcap` | Protocol parsing | <5MB | Various protocols | - -### Generating Test Data - -```python -# scripts/generate_test_pcaps.py -from scapy.all import * - -def create_beacon_pcap(output_path): - """Create PCAP with beaconing pattern.""" - packets = [] - for i in range(100): - pkt = IP(src="192.168.1.100", dst="10.0.0.1") / \ - TCP(sport=RandShort(), dport=443) / \ - Raw(load=b"beacon") - pkt.time = i * 60 # 1 minute intervals - packets.append(pkt) - wrpcap(output_path, packets) -``` - ---- - -## Coverage Requirements - -| Module | Required Coverage | Critical Paths | -|--------|-------------------|----------------| -| `app/utils/export.py` | 95% | All export functions | -| `app/utils/config_manager.py` | 90% | Encryption, load/save | -| `app/pipeline/osint_cache.py` | 90% | Cache operations | -| `app/pipeline/ja3.py` | 85% | Hash calculation | -| `app/pipeline/dns_analysis.py` | 85% | DGA detection | -| `app/pipeline/yara_scan.py` | 80% | Scan functions | -| `app/db/cases.py` | 90% | CRUD operations | - ---- - -## CI/CD Test Configuration - -### pytest.ini - -```ini -[pytest] -testpaths = tests -python_files = test_*.py -python_functions = test_* -addopts = -v --cov=app --cov-report=html --cov-report=term-missing -markers = - slow: marks tests as slow (deselect with '-m "not slow"') - integration: marks integration tests - e2e: marks end-to-end tests -``` - -### GitHub Actions Workflow - -```yaml -name: Tests -on: [push, pull_request] - -jobs: - test: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - name: Set up Python - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - name: Install dependencies - run: | - pip install -e ".[dev]" - sudo apt-get install -y tshark - - name: Run tests - run: pytest -v --cov=app - - name: Upload coverage - uses: codecov/codecov-action@v3 -``` - ---- - -## Test Execution Plan - -### Phase 1 (Quick Wins) -1. Write unit tests for export module -2. Write unit tests for config manager -3. Write unit tests for OSINT cache -4. Write unit tests for JA3 functions - -### Phase 2 (Medium Effort) -1. Write DNS analysis unit tests -2. Write TLS certificate tests -3. Write batch processing integration tests - -### Phase 3 (High Effort) -1. Write YARA scanning tests -2. Write case management unit tests -3. Write case workflow integration tests -4. Write end-to-end tests - ---- - -## Mocking Strategy - -### External APIs - -```python -@pytest.fixture -def mock_virustotal(requests_mock): - requests_mock.get( - re.compile(r"https://www\.virustotal\.com/api/v3/.*"), - json={"data": {"attributes": {"reputation": 0}}} - ) - -@pytest.fixture -def mock_greynoise(requests_mock): - requests_mock.get( - re.compile(r"https://api\.greynoise\.io/v3/community/.*"), - json={"seen": False, "classification": "benign"} - ) -``` - -### File System - -```python -@pytest.fixture -def mock_pcap_read(mocker): - mock_reader = mocker.patch("pyshark.FileCapture") - mock_reader.return_value.__iter__ = lambda self: iter([ - MockPacket(src="192.168.1.1", dst="8.8.8.8") - ]) - return mock_reader -``` - ---- - -## Summary - -This test plan provides comprehensive coverage for all planned features: - -- **Unit Tests**: 40+ test cases covering core functionality -- **Integration Tests**: 10+ test cases for component interactions -- **E2E Tests**: Full workflow validation -- **Coverage Target**: >85% overall, >90% for critical paths -- **Test Data**: Curated PCAP samples for realistic testing diff --git a/docs/zh-TW/README.md b/docs/zh-TW/README.md index 9764547..856ede6 100644 --- a/docs/zh-TW/README.md +++ b/docs/zh-TW/README.md @@ -495,8 +495,6 @@ PCAP Hunter 使用**與生產環境相同形狀的測試資料**，而非簡化 - **[整合 API 參考文件](../API.md)** — REST 端點、認證、設定（英文） - **[API 整合指南（繁體中文）](api/README.md)** — SIEM / SOAR 整合範例 - **[English README](../../README.md)** — 英文版說明 -- **[CLAUDE.md](../../CLAUDE.md)** — 貢獻者 / AI 指南：慣例、測試紀律、已知錯誤模式 -- **[docs/FEATURE-ROADMAP.md](../FEATURE-ROADMAP.md)** — 規劃中的工作 ---