refactor(detmir): retire python runtime paths

aw-server/hayabusa/aw-hayabusa-autoprocess.py

@@ -1,136 +0,0 @@
-#!/usr/bin/env python3
-import argparse
-import fcntl
-import json
-import pathlib
-import shutil
-import subprocess
-import sys
-from typing import Optional
-
-DROP_DIR = pathlib.Path('/opt/activitywatch/aw-rus-ops/drop')
-LOCK_PATH = pathlib.Path('/opt/hayabusa/state/aw-hayabusa-autoprocess.lock')
-WRAPPER = pathlib.Path('/usr/local/bin/aw-hayabusa')
-LINKER = pathlib.Path('/usr/local/bin/aw-hayabusa-link-case')
-CASE_ALERT = pathlib.Path('/usr/local/bin/aw-hayabusa-case-alert')
-
-
-def run(cmd):
-    print('RUN', ' '.join(str(x) for x in cmd), flush=True)
-    subprocess.run(cmd, check=True)
-
-
-def run_capture(cmd):
-    print('RUN', ' '.join(str(x) for x in cmd), flush=True)
-    return subprocess.run(cmd, check=False, text=True, capture_output=True)
-
-
-def read_latest_intake():
-    return json.loads(pathlib.Path('/opt/hayabusa/state/latest-intake.json').read_text(encoding='utf-8'))
-
-
-def load_sidecars(zip_path: pathlib.Path):
-    base = zip_path.with_suffix('')
-    caseid_path = base.with_suffix('.caseid')
-    meta_path = base.with_suffix('.meta.json')
-    meta = {}
-    if meta_path.is_file():
-        meta = json.loads(meta_path.read_text(encoding='utf-8'))
-    case_id = meta.get('case_id')
-    if case_id is None and caseid_path.is_file():
-        raw = caseid_path.read_text(encoding='utf-8').strip()
-        if raw:
-            case_id = int(raw)
-    return {
-        'base': base,
-        'case_id': case_id,
-        'host': meta.get('host'),
-        'mode': meta.get('mode', 'incident'),
-        'link_source': meta.get('link_source', 'aw-rus-drop-autoprocess'),
-        'caseid_path': caseid_path,
-        'meta_path': meta_path,
-    }
-
-
-def archive_sidecars(report_dir: pathlib.Path, sidecars):
-    target_dir = report_dir / 'input-sidecars'
-    target_dir.mkdir(parents=True, exist_ok=True)
-    for path in (sidecars['caseid_path'], sidecars['meta_path']):
-        if path.is_file():
-            shutil.move(str(path), str(target_dir / path.name))
-
-
-def archive_drop_package(report_dir: pathlib.Path, zip_path: pathlib.Path):
-    target_dir = report_dir / 'input-drop'
-    target_dir.mkdir(parents=True, exist_ok=True)
-    target_path = target_dir / zip_path.name
-    if target_path.exists():
-        target_path.unlink()
-    shutil.move(str(zip_path), str(target_path))
-
-
-def guess_host(zip_path: pathlib.Path, sidecars) -> Optional[str]:
-    if sidecars['host']:
-        return str(sidecars['host'])
-    name = zip_path.stem
-    if '-' in name:
-        return name.split('-', 1)[0]
-    return name or None
-
-
-def process_one(zip_path: pathlib.Path):
-    sidecars = load_sidecars(zip_path)
-    host = guess_host(zip_path, sidecars)
-    mode = sidecars['mode'] or 'incident'
-    accept_cmd = [str(WRAPPER), 'accept', '--package', str(zip_path)]
-    if host:
-        accept_cmd += ['--host', host]
-    run(accept_cmd)
-    run([str(WRAPPER), 'process-inbox', '--mode', mode, '--limit', '1'])
-    latest = read_latest_intake()
-    report_dir = pathlib.Path(latest['report_dir'])
-    case_alert = None
-    if CASE_ALERT.is_file():
-        alert_cmd = [str(CASE_ALERT), '--mode', mode, '--link-source', sidecars['link_source']]
-        if sidecars['case_id'] is not None:
-            alert_cmd += ['--case-id', str(sidecars['case_id'])]
-        result = run_capture(alert_cmd)
-        case_alert = {
-            'returncode': result.returncode,
-            'stdout': result.stdout.strip(),
-            'stderr': result.stderr.strip(),
-        }
-    archive_sidecars(report_dir, sidecars)
-    archive_drop_package(report_dir, zip_path)
-    if sidecars['case_id'] is not None and not CASE_ALERT.is_file():
-        run([str(LINKER), '--case-id', str(sidecars['case_id']), '--mode', mode, '--link-source', sidecars['link_source']])
-    return {'latest_intake': latest, 'case_alert': case_alert}
-
-
-def main():
-    p = argparse.ArgumentParser(description='Auto-process Hayabusa zip packages dropped onto aw-rus server')
-    p.add_argument('--drop-dir', default=str(DROP_DIR))
-    p.add_argument('--once', action='store_true', default=True)
-    args = p.parse_args()
-
-    drop_dir = pathlib.Path(args.drop_dir)
-    drop_dir.mkdir(parents=True, exist_ok=True)
-    LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
-    with LOCK_PATH.open('w') as lock_fh:
-        try:
-            fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except BlockingIOError:
-            print('autoprocess already running', file=sys.stderr)
-            return 0
-        zips = sorted(drop_dir.glob('*.zip'))
-        if not zips:
-            print('no zip packages in drop dir')
-            return 0
-        for zip_path in zips:
-            result = process_one(zip_path)
-            print(json.dumps({'processed': str(zip_path), **result}, ensure_ascii=False, indent=2))
-    return 0
-
-
-if __name__ == '__main__':
-    raise SystemExit(main())

aw-server/hayabusa/aw-hayabusa-case-alert.py

@@ -1,361 +0,0 @@
-#!/usr/bin/env python3
-from __future__ import annotations
-
-import argparse
-import csv
-import json
-import os
-import pathlib
-import urllib.error
-import urllib.parse
-import urllib.request
-from collections import Counter
-from datetime import datetime, timezone
-from typing import Any
-
-
-SEVERITY_ORDER = {"low": 1, "medium": 2, "high": 3, "critical": 4}
-LEVEL_NORMALIZATION = {
-    "informational": "info",
-    "info": "info",
-    "low": "low",
-    "med": "med",
-    "medium": "med",
-    "high": "high",
-    "crit": "crit",
-    "critical": "crit",
-}
-LEVEL_WEIGHTS = {
-    "info": 1,
-    "low": 4,
-    "med": 12,
-    "high": 40,
-    "crit": 100,
-}
-
-
-def env_bool(name: str, default: bool) -> bool:
-    value = os.environ.get(name)
-    if value is None:
-        return default
-    return value.strip().lower() in {"1", "true", "yes", "on"}
-
-
-def normalize_level(level: str | None) -> str:
-    if not level:
-        return "info"
-    return LEVEL_NORMALIZATION.get(level.strip().lower(), level.strip().lower())
-
-
-def severity_meets(actual: str, threshold: str) -> bool:
-    return SEVERITY_ORDER.get(actual, 0) >= SEVERITY_ORDER.get(threshold, 0)
-
-
-def build_hayabusa_payload(intake: dict[str, Any], mode: str, link_source: str) -> dict[str, Any]:
-    report_dir = pathlib.Path(intake["report_dir"])
-    return {
-        "tool": "hayabusa",
-        "host": intake["host"],
-        "mode": mode,
-        "status": intake["status"],
-        "intake_id": intake["intake_id"],
-        "package_path": intake["package_path"],
-        "sha256": intake["sha256"],
-        "report_dir": intake["report_dir"],
-        "summary_html": str(report_dir / "summary.html"),
-        "timeline_path": str(report_dir / "timeline.jsonl"),
-        "manifest_path": str(report_dir / "manifest.json"),
-        "link_source": link_source,
-    }
-
-
-def post_json(url: str, payload: dict[str, Any]) -> dict[str, Any]:
-    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
-    req = urllib.request.Request(url, data=data, method="POST", headers={"Content-Type": "application/json"})
-    with urllib.request.urlopen(req) as resp:
-        body = resp.read().decode("utf-8")
-        return json.loads(body) if body else {}
-
-
-def patch_json(url: str, payload: dict[str, Any]) -> dict[str, Any]:
-    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
-    req = urllib.request.Request(url, data=data, method="PATCH", headers={"Content-Type": "application/json"})
-    with urllib.request.urlopen(req) as resp:
-        body = resp.read().decode("utf-8")
-        return json.loads(body) if body else {}
-
-
-def get_json(url: str) -> dict[str, Any]:
-    with urllib.request.urlopen(url) as resp:
-        return json.loads(resp.read().decode("utf-8"))
-
-
-def parse_timestamp(value: str | None) -> datetime | None:
-    if not value:
-        return None
-    try:
-        return datetime.fromisoformat(value.replace("Z", "+00:00"))
-    except ValueError:
-        return None
-
-
-def read_csv_rows(path: pathlib.Path) -> int:
-    if not path.is_file():
-        return 0
-    with path.open("r", encoding="utf-8-sig", newline="") as fh:
-        reader = csv.reader(fh)
-        rows = list(reader)
-    if not rows:
-        return 0
-    return max(0, len(rows) - 1)
-
-
-def analyze_report(report_dir: pathlib.Path) -> dict[str, Any]:
-    timeline_path = report_dir / "timeline.jsonl"
-    level_counts: Counter[str] = Counter()
-    title_counts: Counter[str] = Counter()
-    first_ts: datetime | None = None
-    last_ts: datetime | None = None
-    total_events = 0
-
-    if timeline_path.is_file():
-        with timeline_path.open("r", encoding="utf-8") as fh:
-            for raw_line in fh:
-                line = raw_line.strip()
-                if not line:
-                    continue
-                try:
-                    event = json.loads(line)
-                except json.JSONDecodeError:
-                    continue
-                total_events += 1
-                level = normalize_level(str(event.get("Level", "")))
-                level_counts[level] += 1
-                title = str(event.get("RuleTitle") or "").strip() or "Unknown rule"
-                title_counts[title] += 1
-                ts = parse_timestamp(event.get("Timestamp"))
-                if ts is not None:
-                    first_ts = ts if first_ts is None or ts < first_ts else first_ts
-                    last_ts = ts if last_ts is None or ts > last_ts else last_ts
-
-    failed_logons = read_csv_rows(report_dir / "logon-summary-failed.csv")
-    successful_logons = read_csv_rows(report_dir / "logon-summary-successful.csv")
-
-    suspicious_pwsh = sum(
-        count
-        for title, count in title_counts.items()
-        if "pwsh" in title.lower() or "powershell" in title.lower() or "obfuscation" in title.lower()
-    )
-    credential_events = sum(count for title, count in title_counts.items() if "credential" in title.lower())
-    timestomp_events = sum(count for title, count in title_counts.items() if "timestomp" in title.lower())
-    logon_failure_events = sum(count for title, count in title_counts.items() if "logon failure" in title.lower())
-
-    score = (
-        sum(LEVEL_WEIGHTS.get(level, 0) * count for level, count in level_counts.items())
-        + min(failed_logons, 200) * 2
-        + suspicious_pwsh * 6
-        + credential_events * 8
-        + timestomp_events * 12
-        + logon_failure_events * 2
-    )
-
-    crit_count = level_counts.get("crit", 0)
-    high_count = level_counts.get("high", 0)
-    med_count = level_counts.get("med", 0)
-
-    if crit_count >= 1 or score >= 240 or (high_count >= 4 and suspicious_pwsh >= 4):
-        severity = "critical"
-    elif high_count >= 1 or score >= 120 or suspicious_pwsh >= 8 or credential_events >= 5:
-        severity = "high"
-    elif med_count >= 1 or score >= 40 or failed_logons >= 10:
-        severity = "medium"
-    else:
-        severity = "low"
-
-    return {
-        "severity": severity,
-        "score": score,
-        "events_total": total_events,
-        "level_counts": dict(level_counts),
-        "top_rules": [{"title": title, "count": count} for title, count in title_counts.most_common(5)],
-        "first_timestamp": first_ts.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") if first_ts else None,
-        "last_timestamp": last_ts.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") if last_ts else None,
-        "failed_logon_rows": failed_logons,
-        "successful_logon_rows": successful_logons,
-        "suspicious_pwsh": suspicious_pwsh,
-        "credential_events": credential_events,
-        "timestomp_events": timestomp_events,
-        "logon_failure_events": logon_failure_events,
-    }
-
-
-def build_case_title(host: str, summary: dict[str, Any]) -> str:
-    top_rule = summary.get("top_rules") or []
-    suffix = top_rule[0]["title"] if top_rule else "No dominant rule"
-    return f"Hayabusa {summary['severity'].upper()} · {host} · {suffix}"
-
-
-def build_case_payload(intake: dict[str, Any], summary: dict[str, Any]) -> dict[str, Any]:
-    return {
-        "incident_id": f"hayabusa:{intake['host']}:{intake['intake_id']}",
-        "host": intake["host"],
-        "title": build_case_title(intake["host"], summary),
-        "severity": summary["severity"],
-        "evidence": {
-            "hayabusa": {
-                "intake_id": intake["intake_id"],
-                "package_path": intake["package_path"],
-                "sha256": intake["sha256"],
-                "report_dir": intake["report_dir"],
-                "summary": summary,
-            }
-        },
-    }
-
-
-def build_comment(summary: dict[str, Any], intake: dict[str, Any]) -> str:
-    top = ", ".join(f"{item['title']} ({item['count']})" for item in summary.get("top_rules", [])[:3]) or "n/a"
-    return (
-        f"Hayabusa auto-summary\n"
-        f"Severity: {summary['severity']} (score={summary['score']})\n"
-        f"Host: {intake['host']}\n"
-        f"Intake: {intake['intake_id']}\n"
-        f"Events: {summary['events_total']}, failed_logons={summary['failed_logon_rows']}, "
-        f"suspicious_pwsh={summary['suspicious_pwsh']}, credential_events={summary['credential_events']}\n"
-        f"Top rules: {top}\n"
-        f"Report: {intake['report_dir']}"
-    )
-
-
-def send_telegram(bot_token: str, chat_ids: list[str], text: str) -> list[dict[str, Any]]:
-    results = []
-    for chat_id in chat_ids:
-        payload = urllib.parse.urlencode({"chat_id": chat_id, "text": text}).encode("utf-8")
-        url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
-        req = urllib.request.Request(url, data=payload, method="POST")
-        try:
-            with urllib.request.urlopen(req) as resp:
-                body = json.loads(resp.read().decode("utf-8"))
-            results.append({"chat_id": chat_id, "ok": True, "response": body})
-        except Exception as exc:  # noqa: BLE001
-            results.append({"chat_id": chat_id, "ok": False, "error": str(exc)})
-    return results
-
-
-def build_telegram_text(case_id: int | None, intake: dict[str, Any], summary: dict[str, Any]) -> str:
-    severity_label = {
-        "critical": "критичное событие",
-        "high": "опасное событие",
-        "medium": "подозрительное событие",
-        "low": "слабый сигнал",
-    }.get(summary["severity"], summary["severity"])
-    top_rules = summary.get("top_rules", [])
-    top_rule = top_rules[0]["title"] if top_rules else "нет явного доминирующего правила"
-    lines = [
-        f"Hayabusa: {severity_label}",
-        "",
-        f"Хост: {intake['host']}",
-    ]
-    if case_id is not None:
-        lines.append(f"Кейс: {case_id}")
-    lines.extend(
-        [
-            f"Уровень: {summary['severity']}",
-            "",
-            "Что найдено:",
-            f"- {top_rule}: {top_rules[0]['count'] if top_rules else 0}",
-            f"- подозрительный PowerShell: {summary['suspicious_pwsh']}",
-            f"- ошибок входа: {summary['logon_failure_events']}",
-            f"- событий по учётным данным: {summary['credential_events']}",
-        ]
-    )
-    if summary.get("timestomp_events", 0) > 0:
-        lines.append(f"- timestomp-подобных событий: {summary['timestomp_events']}")
-    lines.extend(
-        [
-            "",
-            "Главный риск:",
-            "возможная активность вокруг учётных данных и PowerShell",
-            "",
-            "Отчёт:",
-            f"{intake['report_dir']}",
-        ]
-    )
-    return "\n".join(lines)
-
-
-def main() -> int:
-    p = argparse.ArgumentParser(description="Auto-create/update AW-rus case, compute Hayabusa severity, and send Telegram alerts")
-    p.add_argument("--case-id", type=int)
-    p.add_argument("--intake-json", default="/opt/hayabusa/state/latest-intake.json")
-    p.add_argument("--case-api-base", default=os.environ.get("AW_HAYABUSA_CASE_API_BASE", "http://127.0.0.1:5602"))
-    p.add_argument("--mode", default="incident")
-    p.add_argument("--link-source", default="aw-rus-drop-autoprocess")
-    p.add_argument("--auto-create", action="store_true", default=env_bool("AW_HAYABUSA_AUTO_CASE_ENABLED", True))
-    p.add_argument("--auto-create-min-severity", default=os.environ.get("AW_HAYABUSA_AUTO_CASE_MIN_SEVERITY", "medium"))
-    p.add_argument("--telegram-enabled", action="store_true", default=env_bool("AW_HAYABUSA_TELEGRAM_ENABLED", False))
-    p.add_argument("--telegram-min-severity", default=os.environ.get("AW_HAYABUSA_TELEGRAM_MIN_SEVERITY", "high"))
-    p.add_argument("--telegram-bot-token", default=os.environ.get("AW_HAYABUSA_TELEGRAM_BOT_TOKEN", ""))
-    p.add_argument("--telegram-chat-ids", default=os.environ.get("AW_HAYABUSA_TELEGRAM_CHAT_IDS", ""))
-    args = p.parse_args()
-
-    intake_path = pathlib.Path(args.intake_json)
-    intake = json.loads(intake_path.read_text(encoding="utf-8"))
-    summary = analyze_report(pathlib.Path(intake["report_dir"]))
-    case_api_base = args.case_api_base.rstrip("/")
-    if case_api_base.endswith("/api/0/dlp/cases"):
-        case_api_base = case_api_base[: -len("/api/0/dlp/cases")]
-    case_id = args.case_id
-    created_case = None
-    case_error = None
-    linked = False
-    comment_added = False
-
-    try:
-        if case_id is None and args.auto_create and severity_meets(summary["severity"], args.auto_create_min_severity):
-            created_case = post_json(f"{case_api_base}/api/0/dlp/cases", build_case_payload(intake, summary))
-            case_id = int(created_case["id"])
-
-        if case_id is not None:
-            patch_json(
-                f"{case_api_base}/api/0/dlp/cases/{case_id}",
-                {"severity": summary["severity"]},
-            )
-            post_json(
-                f"{case_api_base}/api/0/dlp/cases/{case_id}/forensics/hayabusa",
-                build_hayabusa_payload(intake, args.mode, args.link_source),
-            )
-            linked = True
-            post_json(
-                f"{case_api_base}/api/0/dlp/cases/{case_id}/comments",
-                {"comment": build_comment(summary, intake), "author": "aw-hayabusa-auto"},
-            )
-            comment_added = True
-    except Exception as exc:  # noqa: BLE001
-        case_error = str(exc)
-
-    telegram_results: list[dict[str, Any]] = []
-    if args.telegram_enabled and args.telegram_bot_token and severity_meets(summary["severity"], args.telegram_min_severity):
-        chat_ids = [item.strip() for item in args.telegram_chat_ids.split(",") if item.strip()]
-        if chat_ids:
-            telegram_results = send_telegram(
-                bot_token=args.telegram_bot_token,
-                chat_ids=chat_ids,
-                text=build_telegram_text(case_id, intake, summary),
-            )
-
-    result = {
-        "summary": summary,
-        "case_id": case_id,
-        "case_created": created_case,
-        "case_linked": linked,
-        "case_comment_added": comment_added,
-        "case_error": case_error,
-        "telegram_results": telegram_results,
-    }
-    print(json.dumps(result, ensure_ascii=False, indent=2))
-    return 0 if case_error is None else 1
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

aw-server/hayabusa/aw-hayabusa-from-windows.py

@@ -1,111 +0,0 @@
-#!/usr/bin/env python3
-import argparse
-import json
-import pathlib
-import re
-import subprocess
-import sys
-
-OPS_ROOT = pathlib.Path('/opt/activitywatch/aw-rus-ops')
-DEFAULT_INVENTORY = OPS_ROOT / 'ansible' / 'inventory.ini'
-DEFAULT_DROP = OPS_ROOT / 'drop'
-DEFAULT_ANSIBLE = OPS_ROOT / 'venv' / 'bin' / 'ansible'
-DEFAULT_WRAPPER = pathlib.Path('/usr/local/bin/aw-hayabusa')
-DEFAULT_LINKER = pathlib.Path('/usr/local/bin/aw-hayabusa-link-case')
-WINDOWS_EXPORT_CMD = r"powershell.exe -ExecutionPolicy Bypass -File C:\ProgramData\AWatch-rus\export-evtx-for-hayabusa.ps1 -DaysBack {days_back} | ConvertTo-Json -Depth 8 -Compress"
-WINDOWS_LATEST_ZIP_CMD = r"Get-ChildItem 'C:\ProgramData\AWatch-rus\forensics\evtx-exports' -File -Filter '*.zip' | Sort-Object LastWriteTime -Descending | Select-Object -First 1 FullName,Length,LastWriteTime | ConvertTo-Json -Compress"
-
-
-def run(cmd):
-    proc = subprocess.run(cmd, text=True, capture_output=True)
-    if proc.returncode != 0:
-        sys.stderr.write(proc.stdout)
-        sys.stderr.write(proc.stderr)
-        raise SystemExit(proc.returncode)
-    return proc.stdout
-
-
-def extract_json_blob(text):
-    matches = re.findall(r'(\{.*\}|\[.*\])', text, re.S)
-    for candidate in reversed(matches):
-        try:
-            return json.loads(candidate)
-        except Exception:
-            continue
-    raise SystemExit('cannot parse JSON from ansible output:\n' + text)
-
-
-def main():
-    p = argparse.ArgumentParser(description='Run Windows EVTX export and Hayabusa intake directly from aw-server, without the laptop')
-    p.add_argument('--inventory', default=str(DEFAULT_INVENTORY))
-    p.add_argument('--ansible-bin', default=str(DEFAULT_ANSIBLE))
-    p.add_argument('--drop-dir', default=str(DEFAULT_DROP))
-    p.add_argument('--days-back', type=int, default=1)
-    p.add_argument('--mode', default='incident', choices=['quick', 'incident', 'full'])
-    p.add_argument('--case-id', type=int)
-    p.add_argument('--link-source', default='aw-rus-ops-from-windows')
-    p.add_argument('--windows-group', default='aw_windows')
-    p.add_argument('--wrapper', default=str(DEFAULT_WRAPPER))
-    p.add_argument('--linker', default=str(DEFAULT_LINKER))
-    args = p.parse_args()
-
-    inventory = pathlib.Path(args.inventory)
-    ansible_bin = pathlib.Path(args.ansible_bin)
-    drop_dir = pathlib.Path(args.drop_dir)
-    wrapper = pathlib.Path(args.wrapper)
-    linker = pathlib.Path(args.linker)
-    if not inventory.is_file():
-        raise SystemExit(f'inventory not found: {inventory}')
-    if not ansible_bin.is_file():
-        raise SystemExit(f'ansible binary not found: {ansible_bin}')
-    if not wrapper.is_file():
-        raise SystemExit(f'wrapper not found: {wrapper}')
-    drop_dir.mkdir(parents=True, exist_ok=True)
-
-    export_cmd = [str(ansible_bin), args.windows_group, '-i', str(inventory), '-m', 'win_shell', '-a', WINDOWS_EXPORT_CMD.format(days_back=args.days_back)]
-    print('RUN_EXPORT', ' '.join(export_cmd))
-    export_out = run(export_cmd)
-    export_json = extract_json_blob(export_out)
-
-    list_cmd = [str(ansible_bin), args.windows_group, '-i', str(inventory), '-m', 'win_shell', '-a', WINDOWS_LATEST_ZIP_CMD]
-    print('RUN_LIST', ' '.join(list_cmd))
-    latest_out = run(list_cmd)
-    latest = extract_json_blob(latest_out)
-    if isinstance(latest, list):
-        latest = latest[0]
-    remote_zip = latest['FullName']
-    filename = pathlib.PureWindowsPath(remote_zip).name
-    local_zip = drop_dir / filename
-
-    remote_zip_posix = remote_zip.replace('\\', '/')
-    fetch_cmd = [str(ansible_bin), args.windows_group, '-i', str(inventory), '-m', 'fetch', '-a', f'src={remote_zip_posix} dest={drop_dir}/ flat=yes']
-    print('RUN_FETCH', ' '.join(fetch_cmd))
-    run(fetch_cmd)
-    if not local_zip.is_file():
-        raise SystemExit(f'fetched zip not found: {local_zip}')
-
-    accept_cmd = [str(wrapper), 'accept', '--package', str(local_zip)]
-    host = export_json.get('hostname') or pathlib.Path(filename).stem.split('-')[0]
-    if host:
-        accept_cmd += ['--host', str(host)]
-    print('RUN_ACCEPT', ' '.join(accept_cmd))
-    subprocess.run(accept_cmd, check=True)
-
-    process_cmd = [str(wrapper), 'process-inbox', '--mode', args.mode, '--limit', '1']
-    print('RUN_PROCESS', ' '.join(process_cmd))
-    subprocess.run(process_cmd, check=True)
-
-    if args.case_id is not None:
-        if not linker.is_file():
-            raise SystemExit(f'linker not found: {linker}')
-        link_cmd = [str(linker), '--case-id', str(args.case_id), '--mode', args.mode, '--link-source', args.link_source]
-        print('RUN_LINK', ' '.join(link_cmd))
-        subprocess.run(link_cmd, check=True)
-
-    latest_intake = pathlib.Path('/opt/hayabusa/state/latest-intake.json')
-    print('LATEST_INTAKE')
-    print(latest_intake.read_text(encoding='utf-8'))
-
-
-if __name__ == '__main__':
-    main()

aw-server/hayabusa/aw-hayabusa-link-case.py

@@ -1,76 +0,0 @@
-#!/usr/bin/env python3
-import argparse
-import json
-import pathlib
-import sys
-import urllib.error
-import urllib.request
-
-
-def normalize_host(value):
-    return str(value or '').strip().lower()
-
-
-def build_payload(intake, mode, link_source):
-    report_dir = pathlib.Path(intake['report_dir'])
-    return {
-        'tool': 'hayabusa',
-        'host': intake['host'],
-        'mode': mode,
-        'status': intake['status'],
-        'intake_id': intake['intake_id'],
-        'package_path': intake['package_path'],
-        'sha256': intake['sha256'],
-        'report_dir': intake['report_dir'],
-        'summary_html': str(report_dir / 'summary.html'),
-        'timeline_path': str(report_dir / 'timeline.jsonl'),
-        'manifest_path': str(report_dir / 'manifest.json'),
-        'link_source': link_source,
-    }
-
-
-def post_json(url, payload):
-    data = json.dumps(payload, ensure_ascii=False).encode('utf-8')
-    req = urllib.request.Request(url, data=data, method='POST', headers={'Content-Type': 'application/json'})
-    with urllib.request.urlopen(req) as resp:
-        return json.loads(resp.read().decode('utf-8'))
-
-
-def get_json(url):
-    with urllib.request.urlopen(url) as resp:
-        return json.loads(resp.read().decode('utf-8'))
-
-
-def main():
-    p = argparse.ArgumentParser(description='Link latest or specified Hayabusa intake metadata to AW-rus case management')
-    p.add_argument('--case-id', type=int, required=True)
-    p.add_argument('--intake-json', default='/opt/hayabusa/state/latest-intake.json')
-    p.add_argument('--case-api-base', default='http://127.0.0.1:5602')
-    p.add_argument('--mode', default='incident')
-    p.add_argument('--link-source', default='aw-rus-ops')
-    args = p.parse_args()
-
-    intake_path = pathlib.Path(args.intake_json)
-    if not intake_path.is_file():
-        raise SystemExit(f'intake json not found: {intake_path}')
-    intake = json.loads(intake_path.read_text(encoding='utf-8'))
-    case = get_json(f"{args.case_api_base.rstrip('/')}/api/0/dlp/cases/{args.case_id}")
-    case_host = normalize_host(case.get('host'))
-    intake_host = normalize_host(intake.get('host'))
-    if case_host and intake_host and case_host != intake_host:
-        raise SystemExit(
-            f"hayabusa host mismatch: case host={case.get('host')} intake host={intake.get('host')}"
-        )
-    payload = build_payload(intake, args.mode, args.link_source)
-    case_url = f"{args.case_api_base.rstrip('/')}/api/0/dlp/cases/{args.case_id}/forensics/hayabusa"
-    try:
-        post_json(case_url, payload)
-    except urllib.error.HTTPError as exc:
-        body = exc.read().decode('utf-8', errors='replace')
-        raise SystemExit(f'case API POST failed: HTTP {exc.code}: {body}')
-    case = get_json(f"{args.case_api_base.rstrip('/')}/api/0/dlp/cases/{args.case_id}")
-    print(json.dumps({'case_id': args.case_id, 'intake': intake, 'forensics': case.get('forensics')}, ensure_ascii=False, indent=2))
-
-
-if __name__ == '__main__':
-    main()

aw-server/hayabusa/aw-hayabusa.sh

@@ -96,19 +96,8 @@ EOF
 json_field() {
   local json_path="$1"
   local field_name="$2"
-  python3 - "$json_path" "$field_name" <<'PY'
-import json, sys
-path, field = sys.argv[1], sys.argv[2]
-try:
-    with open(path, 'r', encoding='utf-8') as fh:
-        obj = json.load(fh)
-except Exception:
-    sys.exit(0)
-value = obj.get(field)
-if value is None:
-    sys.exit(0)
-print(str(value))
-PY
+  command -v jq >/dev/null 2>&1 || fail "jq is required to read ${json_path}"
+  jq -er --arg field "${field_name}" '.[$field] // empty | tostring' "${json_path}" 2>/dev/null || true
 }
 
 detect_host_from_manifest() {
@@ -124,34 +113,19 @@ detect_host_from_manifest() {
 extract_zip_normalized() {
   local package_path="$1"
   local dest_dir="$2"
-  python3 - "${package_path}" "${dest_dir}" <<'PY'
-import pathlib
-import shutil
-import sys
-import zipfile
-
-zip_path = pathlib.Path(sys.argv[1])
-dest_dir = pathlib.Path(sys.argv[2])
-dest_dir.mkdir(parents=True, exist_ok=True)
-
-with zipfile.ZipFile(zip_path) as zf:
-    for info in zf.infolist():
-        raw_name = info.filename.replace('\\', '/')
-        normalized = pathlib.PurePosixPath(raw_name)
-        parts = [part for part in normalized.parts if part not in ('', '.')]
-        if any(part == '..' for part in parts):
-            raise SystemExit(f'unsafe zip entry: {info.filename}')
-        if not parts:
-            continue
-        target = dest_dir.joinpath(*parts)
-        is_dir = info.is_dir() or raw_name.endswith('/')
-        if is_dir:
-            target.mkdir(parents=True, exist_ok=True)
-            continue
-        target.parent.mkdir(parents=True, exist_ok=True)
-        with zf.open(info) as src, target.open('wb') as dst:
-            shutil.copyfileobj(src, dst)
-PY
+  command -v zipinfo >/dev/null 2>&1 || fail "zipinfo is required to inspect ${package_path}"
+  command -v unzip >/dev/null 2>&1 || fail "unzip is required to extract ${package_path}"
+  mkdir -p "${dest_dir}"
+  local entry normalized
+  while IFS= read -r entry; do
+    normalized="${entry//\\//}"
+    case "${normalized}" in
+      ""|.|/*|*"/../"*|../*|*".."|*"/..")
+        fail "unsafe zip entry: ${entry}"
+        ;;
+    esac
+  done < <(zipinfo -1 "${package_path}")
+  unzip -q "${package_path}" -d "${dest_dir}"
 }
 
 write_package_manifest() {