up

tools/certbund_offline_snapshot.py

@@ -0,0 +1,444 @@
+#!/usr/bin/env python3
+"""
+Capture CERT-Bund search/export JSON snapshots and generate Offline Kit manifests.
+
+The script can bootstrap a session against https://wid.cert-bund.de, fetch
+paginated search results plus per-year export payloads, and emit a manifest
+that records source, date range, SHA-256, and capture timestamps for each artefact.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import hashlib
+import json
+import os
+from pathlib import Path, PurePosixPath
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from http.cookiejar import MozillaCookieJar
+from typing import Any, Dict, Iterable, List, Optional
+
+
+PORTAL_ROOT = "https://wid.cert-bund.de/portal/"
+SEARCH_ENDPOINT = "https://wid.cert-bund.de/portal/api/securityadvisory/search"
+EXPORT_ENDPOINT = "https://wid.cert-bund.de/portal/api/securityadvisory/export"
+CSRF_ENDPOINT = "https://wid.cert-bund.de/portal/api/security/csrf"
+USER_AGENT = "StellaOps.CertBundOffline/0.1"
+
+UTC = dt.timezone.utc
+
+
+class CertBundClient:
+    def __init__(
+        self,
+        cookie_file: Optional[Path] = None,
+        xsrf_token: Optional[str] = None,
+        auto_bootstrap: bool = True,
+    ) -> None:
+        self.cookie_path = cookie_file
+        self.cookie_jar = MozillaCookieJar()
+
+        if self.cookie_path and self.cookie_path.exists():
+            self.cookie_jar.load(self.cookie_path, ignore_discard=True, ignore_expires=True)
+
+        self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cookie_jar))
+        self.opener.addheaders = [("User-Agent", USER_AGENT)]
+
+        self._xsrf_token = xsrf_token
+        self.auto_bootstrap = auto_bootstrap
+
+        if self.auto_bootstrap and not self._xsrf_token:
+            self._bootstrap()
+
+    @property
+    def xsrf_token(self) -> str:
+        if self._xsrf_token:
+            return self._xsrf_token
+
+        token = _extract_cookie_value(self.cookie_jar, "XSRF-TOKEN")
+        if token:
+            self._xsrf_token = token
+            return token
+
+        raise RuntimeError(
+            "CERT-Bund XSRF token not available. Provide --xsrf-token or a cookie file "
+            "containing XSRF-TOKEN (see docs/ops/feedser-certbund-operations.md)."
+        )
+
+    def fetch_search_pages(
+        self,
+        destination: Path,
+        page_size: int,
+        max_pages: int,
+    ) -> None:
+        destination.mkdir(parents=True, exist_ok=True)
+
+        for page in range(max_pages):
+            payload = {
+                "page": page,
+                "size": page_size,
+                "sort": ["published,desc"],
+            }
+            try:
+                document = self._post_json(SEARCH_ENDPOINT, payload)
+            except urllib.error.HTTPError as exc:
+                raise RuntimeError(
+                    f"Failed to fetch CERT-Bund search page {page}: HTTP {exc.code}. "
+                    "Double-check the XSRF token or portal cookies."
+                ) from exc
+
+            content = document.get("content") or []
+            if not content and page > 0:
+                break
+
+            file_path = destination / f"certbund-search-page-{page:02d}.json"
+            _write_pretty_json(file_path, document)
+            print(f"[certbund] wrote search page {page:02d} → {file_path}")
+
+            if not content:
+                break
+
+        self._persist_cookies()
+
+    def fetch_exports(self, destination: Path, start_year: int, end_year: int) -> None:
+        destination.mkdir(parents=True, exist_ok=True)
+
+        for year in range(start_year, end_year + 1):
+            from_value = f"{year}-01-01"
+            to_value = f"{year}-12-31"
+            query = urllib.parse.urlencode({"format": "json", "from": from_value, "to": to_value})
+            url = f"{EXPORT_ENDPOINT}?{query}"
+            try:
+                document = self._get_json(url)
+            except urllib.error.HTTPError as exc:
+                raise RuntimeError(
+                    f"Failed to fetch CERT-Bund export for {year}: HTTP {exc.code}. "
+                    "Ensure the XSRF token and cookies are valid."
+                ) from exc
+
+            file_path = destination / f"certbund-export-{year}.json"
+            _write_pretty_json(file_path, document)
+            print(f"[certbund] wrote export {year} → {file_path}")
+
+        self._persist_cookies()
+
+    def _bootstrap(self) -> None:
+        try:
+            self._request("GET", PORTAL_ROOT, headers={"Accept": "text/html,application/xhtml+xml"})
+        except urllib.error.HTTPError as exc:
+            raise RuntimeError(f"Failed to bootstrap CERT-Bund session: HTTP {exc.code}") from exc
+
+        # First attempt to obtain CSRF token directly.
+        self._attempt_csrf_fetch()
+
+        if _extract_cookie_value(self.cookie_jar, "XSRF-TOKEN"):
+            return
+
+        # If the token is still missing, trigger the search endpoint once (likely 403)
+        # to make the portal materialise JSESSIONID, then retry token acquisition.
+        try:
+            payload = {"page": 0, "size": 1, "sort": ["published,desc"]}
+            self._post_json(SEARCH_ENDPOINT, payload, include_token=False)
+        except urllib.error.HTTPError:
+            pass
+
+        self._attempt_csrf_fetch()
+
+        token = _extract_cookie_value(self.cookie_jar, "XSRF-TOKEN")
+        if token:
+            self._xsrf_token = token
+        else:
+            print(
+                "[certbund] warning: automatic XSRF token retrieval failed. "
+                "Supply --xsrf-token or reuse a browser-exported cookies file.",
+                file=sys.stderr,
+            )
+
+    def _attempt_csrf_fetch(self) -> None:
+        headers = {
+            "Accept": "application/json, text/plain, */*",
+            "X-Requested-With": "XMLHttpRequest",
+            "Origin": "https://wid.cert-bund.de",
+            "Referer": PORTAL_ROOT,
+        }
+        try:
+            self._request("GET", CSRF_ENDPOINT, headers=headers)
+        except urllib.error.HTTPError:
+            pass
+
+    def _request(self, method: str, url: str, data: Optional[bytes] = None, headers: Optional[Dict[str, str]] = None) -> bytes:
+        request = urllib.request.Request(url, data=data, method=method)
+        default_headers = {
+            "User-Agent": USER_AGENT,
+            "Accept": "application/json",
+        }
+        for key, value in default_headers.items():
+            request.add_header(key, value)
+
+        if headers:
+            for key, value in headers.items():
+                request.add_header(key, value)
+
+        return self.opener.open(request, timeout=60).read()
+
+    def _post_json(self, url: str, payload: Dict[str, Any], include_token: bool = True) -> Dict[str, Any]:
+        data = json.dumps(payload).encode("utf-8")
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "X-Requested-With": "XMLHttpRequest",
+            "Origin": "https://wid.cert-bund.de",
+            "Referer": PORTAL_ROOT,
+        }
+        if include_token:
+            headers["X-XSRF-TOKEN"] = self.xsrf_token
+
+        raw = self._request("POST", url, data=data, headers=headers)
+        return json.loads(raw.decode("utf-8"))
+
+    def _get_json(self, url: str) -> Any:
+        headers = {
+            "Accept": "application/json",
+            "X-Requested-With": "XMLHttpRequest",
+            "Referer": PORTAL_ROOT,
+        }
+        headers["X-XSRF-TOKEN"] = self.xsrf_token
+
+        raw = self._request("GET", url, headers=headers)
+        return json.loads(raw.decode("utf-8"))
+
+    def _persist_cookies(self) -> None:
+        if not self.cookie_path:
+            return
+
+        self.cookie_path.parent.mkdir(parents=True, exist_ok=True)
+        self.cookie_jar.save(self.cookie_path, ignore_discard=True, ignore_expires=True)
+
+
+def _extract_cookie_value(jar: MozillaCookieJar, name: str) -> Optional[str]:
+    for cookie in jar:
+        if cookie.name == name:
+            return cookie.value
+    return None
+
+
+def _write_pretty_json(path: Path, document: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as handle:
+        json.dump(document, handle, ensure_ascii=False, indent=2, sort_keys=True)
+        handle.write("\n")
+
+
+def scan_artifacts(root: Path) -> List[Dict[str, Any]]:
+    records: List[Dict[str, Any]] = []
+    search_dir = root / "search"
+    export_dir = root / "export"
+
+    if search_dir.exists():
+        for file_path in sorted(search_dir.glob("certbund-search-page-*.json")):
+            record = _build_search_record(file_path)
+            records.append(record)
+
+    if export_dir.exists():
+        for file_path in sorted(export_dir.glob("certbund-export-*.json")):
+            record = _build_export_record(file_path)
+            records.append(record)
+
+    return records
+
+
+def _build_search_record(path: Path) -> Dict[str, Any]:
+    with path.open("r", encoding="utf-8") as handle:
+        data = json.load(handle)
+
+    content = data.get("content") or []
+    published_values: List[str] = []
+    for item in content:
+        published = (
+            item.get("published")
+            or item.get("publishedAt")
+            or item.get("datePublished")
+            or item.get("published_date")
+        )
+        if isinstance(published, str):
+            published_values.append(published)
+
+    if published_values:
+        try:
+            ordered = sorted(_parse_iso_timestamp(value) for value in published_values if value)
+            range_from = ordered[0].isoformat()
+            range_to = ordered[-1].isoformat()
+        except ValueError:
+            range_from = range_to = None
+    else:
+        range_from = range_to = None
+
+    return {
+        "type": "search",
+        "path": path,
+        "source": "feedser.cert-bund.search",
+        "itemCount": len(content),
+        "from": range_from,
+        "to": range_to,
+        "capturedAt": _timestamp_from_stat(path),
+    }
+
+
+def _build_export_record(path: Path) -> Dict[str, Any]:
+    year = _extract_year_from_filename(path.name)
+    if year is not None:
+        from_value = f"{year}-01-01"
+        to_value = f"{year}-12-31"
+    else:
+        from_value = None
+        to_value = None
+
+    return {
+        "type": "export",
+        "path": path,
+        "source": "feedser.cert-bund.export",
+        "itemCount": None,
+        "from": from_value,
+        "to": to_value,
+        "capturedAt": _timestamp_from_stat(path),
+    }
+
+
+def _timestamp_from_stat(path: Path) -> str:
+    stat = path.stat()
+    return dt.datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat()
+
+
+def _extract_year_from_filename(name: str) -> Optional[int]:
+    stem = Path(name).stem
+    parts = stem.split("-")
+    if parts and parts[-1].isdigit() and len(parts[-1]) == 4:
+        return int(parts[-1])
+    return None
+
+
+def _parse_iso_timestamp(value: str) -> dt.datetime:
+    try:
+        return dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        # Fallback for formats like 2025-10-14T06:24:49
+        return dt.datetime.strptime(value, "%Y-%m-%dT%H:%M:%S").replace(tzinfo=UTC)
+
+
+def build_manifest(root: Path, records: Iterable[Dict[str, Any]], manifest_path: Path) -> None:
+    manifest_entries = []
+    for record in records:
+        path = record["path"]
+        rel_path = PurePosixPath(path.relative_to(root).as_posix())
+        sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
+        size = path.stat().st_size
+
+        entry = {
+            "source": record["source"],
+            "type": record["type"],
+            "path": str(rel_path),
+            "sha256": sha256,
+            "sizeBytes": size,
+            "capturedAt": record["capturedAt"],
+            "from": record.get("from"),
+            "to": record.get("to"),
+            "itemCount": record.get("itemCount"),
+        }
+        manifest_entries.append(entry)
+
+        sha_file = path.with_suffix(path.suffix + ".sha256")
+        _write_sha_file(sha_file, sha256, path.name)
+
+    manifest_entries.sort(key=lambda item: item["path"])
+
+    manifest_path.parent.mkdir(parents=True, exist_ok=True)
+    manifest_document = {
+        "source": "feedser.cert-bund",
+        "generatedAt": dt.datetime.now(tz=UTC).isoformat(),
+        "artifacts": manifest_entries,
+    }
+
+    with manifest_path.open("w", encoding="utf-8") as handle:
+        json.dump(manifest_document, handle, ensure_ascii=False, indent=2, sort_keys=True)
+        handle.write("\n")
+
+    manifest_sha = hashlib.sha256(manifest_path.read_bytes()).hexdigest()
+    _write_sha_file(manifest_path.with_suffix(".sha256"), manifest_sha, manifest_path.name)
+
+    print(f"[certbund] manifest generated → {manifest_path}")
+
+
+def _write_sha_file(path: Path, digest: str, filename: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as handle:
+        handle.write(f"{digest}  {filename}\n")
+
+
+def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Capture CERT-Bund search/export snapshots for Offline Kit packaging.",
+    )
+    parser.add_argument("--output", default="seed-data/cert-bund", help="Destination directory for artefacts.")
+    parser.add_argument("--start-year", type=int, default=2014, help="First year (inclusive) for export snapshots.")
+    parser.add_argument(
+        "--end-year",
+        type=int,
+        default=dt.datetime.now(tz=UTC).year,
+        help="Last year (inclusive) for export snapshots.",
+    )
+    parser.add_argument("--page-size", type=int, default=100, help="Search page size.")
+    parser.add_argument("--max-pages", type=int, default=12, help="Maximum number of search result pages to capture.")
+    parser.add_argument("--cookie-file", type=Path, help="Path to a Netscape cookie file to reuse/persist session cookies.")
+    parser.add_argument("--xsrf-token", help="Optional explicit XSRF token value (overrides cookie discovery).")
+    parser.add_argument(
+        "--skip-fetch",
+        action="store_true",
+        help="Skip HTTP fetches and only regenerate manifest from existing files.",
+    )
+    parser.add_argument(
+        "--no-bootstrap",
+        action="store_true",
+        help="Do not attempt automatic session bootstrap (use with --skip-fetch or pre-populated cookies).",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    args = parse_args(argv)
+    output_dir = Path(args.output).expanduser().resolve()
+
+    if not args.skip_fetch:
+        client = CertBundClient(
+            cookie_file=args.cookie_file,
+            xsrf_token=args.xsrf_token,
+            auto_bootstrap=not args.no_bootstrap,
+        )
+
+        start_year = args.start_year
+        end_year = args.end_year
+        if start_year > end_year:
+            raise SystemExit("start-year cannot be greater than end-year.")
+
+        client.fetch_search_pages(output_dir / "search", args.page_size, args.max_pages)
+        client.fetch_exports(output_dir / "export", start_year, end_year)
+
+    records = scan_artifacts(output_dir)
+    if not records:
+        print(
+            "[certbund] no artefacts discovered. Fetch data first or point --output to the dataset directory.",
+            file=sys.stderr,
+        )
+        return 1
+
+    manifest_path = output_dir / "manifest" / "certbund-offline-manifest.json"
+    build_manifest(output_dir, records, manifest_path)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())