From 5136feeef73964be20ad7a1eb04a5e3c26880469 Mon Sep 17 00:00:00 2001 From: Naitik Verma Date: Wed, 15 Jul 2026 17:04:38 +0000 Subject: [PATCH 1/2] fix(backend): standardize timezone handling to UTC ISO-8601 Add shared time_utils helpers and use timezone-aware UTC with an explicit offset for generated_at and discovered_at across reports, findings API responses, and report generation. Closes #1882 --- backend/secuscan/executor.py | 37 +++--- backend/secuscan/finding_intelligence.py | 16 ++- backend/secuscan/reporting.py | 15 +-- backend/secuscan/routes.py | 20 +++- backend/secuscan/routes_json_helpers.py | 7 ++ backend/secuscan/time_utils.py | 83 +++++++++++++ frontend/testing/unit/utils/date.test.ts | 11 ++ testing/backend/unit/test_time_utils.py | 145 +++++++++++++++++++++++ 8 files changed, 293 insertions(+), 41 deletions(-) create mode 100644 backend/secuscan/time_utils.py create mode 100644 testing/backend/unit/test_time_utils.py diff --git a/backend/secuscan/executor.py b/backend/secuscan/executor.py index a8523a245..c25ce37ab 100644 --- a/backend/secuscan/executor.py +++ b/backend/secuscan/executor.py @@ -28,6 +28,7 @@ from .models import NotificationDeliveryStatus, TaskStatus, ScanPhase from .ratelimit import concurrent_limiter from .risk_scoring import compute_risk_score, compute_risk_factors +from .time_utils import to_utc_iso from .capabilities import CapabilityEnforcer, CapabilityDeniedError, build_enforcer_from_settings from .parser_sandbox import run_parser_in_sandbox, ParserSandboxError from .network_policy import get_policy_engine @@ -90,17 +91,11 @@ async def _terminate_process_group(pid: int, task_id: str, grace_seconds: int = def _parse_discovered_at(finding: dict) -> Optional[datetime]: - """Extract and parse discovered_at from a finding dict, or return current UTC time.""" - raw = finding.get("discovered_at") - if raw: - try: - if isinstance(raw, str): - return datetime.fromisoformat(raw) - if isinstance(raw, datetime): - return raw - except (ValueError, TypeError): - pass - return datetime.now(timezone.utc) + """Extract and parse discovered_at from a finding dict as timezone-aware UTC.""" + from .time_utils import parse_to_utc, utc_now + + parsed = parse_to_utc(finding.get("discovered_at")) + return parsed if parsed is not None else utc_now() def _validate_risk_fields(finding: dict) -> None: @@ -1362,8 +1357,8 @@ async def _persist_finding( asset_refs = finding.get("asset_refs", []) if isinstance(finding.get("asset_refs"), list) else [] references = finding.get("references", []) if isinstance(finding.get("references"), list) else [] corroborating_sources = finding.get("corroborating_sources", []) if isinstance(finding.get("corroborating_sources"), list) else [] - first_seen_at = str(finding.get("first_seen_at") or discovered.isoformat()) - last_seen_at = str(finding.get("last_seen_at") or discovered.isoformat()) + first_seen_at = str(finding.get("first_seen_at") or to_utc_iso(discovered)) + last_seen_at = str(finding.get("last_seen_at") or to_utc_iso(discovered)) occurrence_count = int(finding.get("occurrence_count") or 1) evidence_count = int(finding.get("evidence_count") or len(evidence)) risk_score = compute_risk_score( @@ -1447,7 +1442,7 @@ async def _persist_finding( finding.get("cvss"), finding.get("cve"), json.dumps(metadata), - discovered.isoformat(), + to_utc_iso(discovered), exploitability, confidence, 1 if finding.get("validated") else 0, @@ -1485,7 +1480,7 @@ async def _persist_finding( "id": finding_id, "plugin_id": plugin_id, "target": target_value, - "discovered_at": discovered.isoformat(), + "discovered_at": to_utc_iso(discovered), "metadata": metadata, "evidence": evidence, "asset_refs": asset_refs, @@ -1539,11 +1534,12 @@ async def _upsert_findings_and_report(self, db, task_id: str, owner_id: str, plu """ INSERT INTO reports ( id, owner_id, task_id, name, type, generated_at, status, findings, pages - ) VALUES (?, ?, ?, ?, ?, (datetime('now')), ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, findings = EXCLUDED.findings, - pages = EXCLUDED.pages + pages = EXCLUDED.pages, + generated_at = EXCLUDED.generated_at """, ( f"report:{task_id}", @@ -1551,6 +1547,7 @@ async def _upsert_findings_and_report(self, db, task_id: str, owner_id: str, plu task_id, f"{plugin.name} Report", "technical", + to_utc_iso(), "ready" if status == TaskStatus.COMPLETED.value else "failed", len(findings_data), 1, @@ -1606,11 +1603,12 @@ async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, owner """ INSERT INTO reports ( id, owner_id, task_id, name, type, generated_at, status, findings, pages - ) VALUES (?, ?, ?, ?, ?, (datetime('now')), ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, findings = EXCLUDED.findings, - pages = EXCLUDED.pages + pages = EXCLUDED.pages, + generated_at = EXCLUDED.generated_at """, ( f"report:{task_id}", @@ -1618,6 +1616,7 @@ async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, owner task_id, f"{scanner.name} Report", "professional" if status == TaskStatus.COMPLETED.value else "failed", + to_utc_iso(), "ready" if status == TaskStatus.COMPLETED.value else "failed", len(findings_data), 2, # Professional reports are typically multi-page diff --git a/backend/secuscan/finding_intelligence.py b/backend/secuscan/finding_intelligence.py index 08a26015f..73c6f2ef1 100644 --- a/backend/secuscan/finding_intelligence.py +++ b/backend/secuscan/finding_intelligence.py @@ -43,7 +43,9 @@ def _now_iso() -> str: - return datetime.now(timezone.utc).isoformat() + from .time_utils import to_utc_iso + + return to_utc_iso() def generate_finding_key(finding: Dict[str, Any], plugin_id: str, target: str, owner_id: str) -> str: @@ -60,14 +62,10 @@ def generate_finding_key(finding: Dict[str, Any], plugin_id: str, target: str, o def _parse_timestamp(raw: Any) -> str: - if isinstance(raw, datetime): - return raw.astimezone(timezone.utc).isoformat() - if isinstance(raw, str) and raw.strip(): - try: - return datetime.fromisoformat(raw.replace("Z", "+00:00")).astimezone(timezone.utc).isoformat() - except ValueError: - return _now_iso() - return _now_iso() + from .time_utils import parse_to_utc, to_utc_iso + + parsed = parse_to_utc(raw) + return to_utc_iso(parsed) if parsed is not None else to_utc_iso() def _stable_id(prefix: str, *parts: Any) -> str: diff --git a/backend/secuscan/reporting.py b/backend/secuscan/reporting.py index 5f85e8e55..f1c370705 100644 --- a/backend/secuscan/reporting.py +++ b/backend/secuscan/reporting.py @@ -15,6 +15,8 @@ from PIL import Image, ImageDraw from xhtml2pdf import pisa +from .time_utils import format_utc_display, to_utc_iso + class ReportGenerator: """Handles PDF, HTML, and CSV generation for security audits.""" @@ -215,7 +217,7 @@ def _normalize_finding(cls, finding: Any) -> Dict[str, Any]: "confidence_reason": redact(cls._clean_text(finding.get("confidence_reason"))), "service_fingerprint": cls._clean_text(finding.get("service_fingerprint")), "cpe": cls._clean_text(finding.get("cpe")), - "discovered_at": cls._clean_text(finding.get("discovered_at")), + "discovered_at": to_utc_iso(finding.get("discovered_at")) if finding.get("discovered_at") else "", "evidence": finding.get("evidence", []) if isinstance(finding.get("evidence"), list) else [], "asset_refs": finding.get("asset_refs", []) if isinstance(finding.get("asset_refs"), list) else [], "references": finding.get("references", []) if isinstance(finding.get("references"), list) else [], @@ -372,8 +374,8 @@ def _build_report_payload(cls, task: Dict[str, Any], result: Dict[str, Any]) -> "tool_name": cls._clean_text(task.get("tool_name")) or cls._clean_text(task.get("plugin_id")) or "Unknown tool", "target": cls._clean_text(task.get("target")) or "Unknown target", "status": cls._clean_text(task.get("status")) or "unknown", - "created_at": cls._clean_text(task.get("created_at")), - "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"), + "created_at": to_utc_iso(task.get("created_at")) if task.get("created_at") else "", + "generated_at": to_utc_iso(), "preset": cls._clean_text(task.get("preset")), "findings": findings, "summary": summary, @@ -389,12 +391,7 @@ def _build_report_payload(cls, task: Dict[str, Any], result: Dict[str, Any]) -> def _format_timestamp(value: str) -> str: if not value: return "Unknown" - for fmt in ("%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"): - try: - return datetime.strptime(value.replace("Z", ""), fmt).strftime("%b %d, %Y %H:%M") - except ValueError: - continue - return value + return format_utc_display(value) @classmethod def _generate_pdf_html_report(cls, task: Dict[str, Any], result: Dict[str, Any]) -> str: diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index f83452d30..a1e48fded 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -1247,12 +1247,20 @@ async def get_reports(owner: str = Depends(get_current_owner)): """Return the caller's generated reports.""" async def build(): + from .time_utils import to_utc_iso + db = await get_db() rows = await db.fetchall( "SELECT * FROM reports WHERE owner_id = ? ORDER BY generated_at DESC", (owner,), ) - return {"reports": parse_json_fields(rows, ["metadata_json"])} + reports = [] + for row in parse_json_fields(rows, ["metadata_json"]): + report = dict(row) + if report.get("generated_at"): + report["generated_at"] = to_utc_iso(report["generated_at"]) + reports.append(report) + return {"reports": reports} return await get_or_set_cached(f"reports:list:{owner}", build) @@ -1269,6 +1277,8 @@ async def search( owner: str = Depends(get_current_owner), ): """Search the caller's findings and reports by title/description/name.""" + from .time_utils import to_utc_iso + db = await get_db() pattern = f"%{_escape_like(q.strip())}%" @@ -1304,7 +1314,7 @@ async def search( "category": row["category"], "severity": row["severity"], "target": row["target"], - "discovered_at": row["discovered_at"], + "discovered_at": to_utc_iso(row["discovered_at"]) if row["discovered_at"] else row["discovered_at"], } for row in finding_rows ], @@ -1314,7 +1324,7 @@ async def search( "task_id": row["task_id"], "name": row["name"], "type": row["type"], - "generated_at": row["generated_at"], + "generated_at": to_utc_iso(row["generated_at"]) if row["generated_at"] else row["generated_at"], } for row in report_rows ], @@ -2581,6 +2591,8 @@ async def list_notification_history( @router.get("/finding/{finding_id}") async def get_finding_details(finding_id: str, owner: str = Depends(get_current_owner)): """Get detailed information for a specific finding""" + from .time_utils import to_utc_iso + db = await get_db() finding_row = await db.fetchone( @@ -2627,7 +2639,7 @@ async def get_finding_details(finding_id: str, owner: str = Depends(get_current_ "proof": finding_row["proof"], "cvss": finding_row["cvss"], "cve": finding_row["cve"], - "discovered_at": finding_row["discovered_at"], + "discovered_at": to_utc_iso(finding_row["discovered_at"]) if finding_row.get("discovered_at") else finding_row.get("discovered_at"), "metadata": metadata, "exploitability": finding_row.get("exploitability"), "confidence": finding_row.get("confidence"), diff --git a/backend/secuscan/routes_json_helpers.py b/backend/secuscan/routes_json_helpers.py index 3d5a7aab4..a18ad12e8 100644 --- a/backend/secuscan/routes_json_helpers.py +++ b/backend/secuscan/routes_json_helpers.py @@ -62,7 +62,10 @@ def deserialize_finding_rows(rows: List[Dict]) -> List[Dict[str, Any]]: The ``*_json`` suffix is stripped from the parsed values: ``metadata_json`` -> ``metadata``, ``evidence_json`` -> ``evidence``, etc. Rows that do not contain a given ``*_json`` key are passed through. + Timestamps are normalized to ISO-8601 UTC with an explicit offset. """ + from .time_utils import to_utc_iso + findings = parse_json_fields(rows, FINDING_JSON_FIELDS) for finding in findings: if "metadata_json" in finding: @@ -78,6 +81,10 @@ def deserialize_finding_rows(rows: List[Dict]) -> List[Dict[str, Any]]: if "corroborating_sources_json" in finding: finding["corroborating_sources"] = finding.pop("corroborating_sources_json") + for ts_field in ("discovered_at", "first_seen_at", "last_seen_at"): + if finding.get(ts_field): + finding[ts_field] = to_utc_iso(finding[ts_field]) + # Expose remediation safety fields at the top level metadata = finding.get("metadata") if isinstance(metadata, dict): diff --git a/backend/secuscan/time_utils.py b/backend/secuscan/time_utils.py new file mode 100644 index 000000000..63fbe7fff --- /dev/null +++ b/backend/secuscan/time_utils.py @@ -0,0 +1,83 @@ +"""Timezone helpers: canonical UTC-aware timestamps for API + reports.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Optional + + +UTC = timezone.utc + + +def utc_now() -> datetime: + """Return the current time as a timezone-aware UTC datetime.""" + return datetime.now(UTC) + + +def ensure_utc(value: datetime) -> datetime: + """Coerce a datetime to timezone-aware UTC. + + Naive datetimes are treated as UTC (matching SQLite ``datetime('now')`` + storage semantics in this project). + """ + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def parse_to_utc(value: Any) -> Optional[datetime]: + """Parse a datetime or ISO-8601 string into timezone-aware UTC. + + Returns ``None`` when the value cannot be parsed. + """ + if value is None: + return None + if isinstance(value, datetime): + return ensure_utc(value) + if isinstance(value, (int, float)): + return datetime.fromtimestamp(value, tz=UTC) + + text = str(value).strip() + if not text or text.lower() == "now": + return utc_now() if text.lower() == "now" else None + + # SQLite stores "YYYY-MM-DD HH:MM:SS" (UTC, no offset) + candidate = text.replace(" ", "T") if "T" not in text else text + try: + parsed = datetime.fromisoformat(candidate.replace("Z", "+00:00")) + except ValueError: + for fmt in ( + "%Y-%m-%dT%H:%M:%S.%f", + "%Y-%m-%dT%H:%M:%S", + "%Y-%m-%d %H:%M:%S.%f", + "%Y-%m-%d %H:%M:%S", + "%Y-%m-%d", + ): + try: + parsed = datetime.strptime(text, fmt) + break + except ValueError: + continue + else: + return None + return ensure_utc(parsed) + + +def to_utc_iso(value: Any = None, *, timespec: str = "seconds") -> str: + """Convert a value to an ISO-8601 UTC string with an explicit offset. + + ``None`` (or omitted) uses the current UTC time. Unparseable values fall + back to ``utc_now()`` so API responses always include a timezone offset. + """ + parsed = utc_now() if value is None else parse_to_utc(value) + if parsed is None: + parsed = utc_now() + return parsed.isoformat(timespec=timespec) + + +def format_utc_display(value: Any, *, fmt: str = "%b %d, %Y %H:%M UTC") -> str: + """Format a timestamp for human-readable report pages in UTC.""" + parsed = parse_to_utc(value) + if parsed is None: + return "Unknown" + return parsed.strftime(fmt) diff --git a/frontend/testing/unit/utils/date.test.ts b/frontend/testing/unit/utils/date.test.ts index 75119f261..467e1ce30 100644 --- a/frontend/testing/unit/utils/date.test.ts +++ b/frontend/testing/unit/utils/date.test.ts @@ -26,6 +26,17 @@ describe("date utilities", () => { expect(result).not.toBeNull() }) + test("treats naive timestamps as UTC for consistent generated_at/discovered_at rendering", () => { + const withZ = parseDateSafe("2026-07-15T12:00:00Z") + const withOffset = parseDateSafe("2026-07-15T12:00:00+00:00") + const naiveIso = parseDateSafe("2026-07-15T12:00:00") + const sqlite = parseDateSafe("2026-07-15 12:00:00") + + expect(withZ?.getTime()).toBe(withOffset?.getTime()) + expect(naiveIso?.getTime()).toBe(withZ?.getTime()) + expect(sqlite?.getTime()).toBe(withZ?.getTime()) + }) + test("returns null for invalid input", () => { expect(parseDateSafe("invalid-date")).toBeNull() }) diff --git a/testing/backend/unit/test_time_utils.py b/testing/backend/unit/test_time_utils.py new file mode 100644 index 000000000..8956e3dbb --- /dev/null +++ b/testing/backend/unit/test_time_utils.py @@ -0,0 +1,145 @@ +"""Tests for UTC timestamp helpers and report timezone standardization.""" + +from datetime import datetime, timezone, timedelta + +import pytest + +from backend.secuscan.time_utils import ( + ensure_utc, + format_utc_display, + parse_to_utc, + to_utc_iso, + utc_now, +) +from backend.secuscan.reporting import ReportGenerator +from backend.secuscan.executor import _parse_discovered_at +from backend.secuscan.routes_json_helpers import deserialize_finding_rows + + +def test_utc_now_is_timezone_aware(): + now = utc_now() + assert now.tzinfo is not None + assert now.utcoffset() == timedelta(0) + + +def test_ensure_utc_treats_naive_as_utc(): + naive = datetime(2026, 7, 15, 12, 0, 0) + aware = ensure_utc(naive) + assert aware.tzinfo == timezone.utc + assert aware.hour == 12 + + +def test_ensure_utc_converts_other_offsets(): + eastern = datetime(2026, 7, 15, 8, 0, 0, tzinfo=timezone(timedelta(hours=-4))) + utc_value = ensure_utc(eastern) + assert utc_value.hour == 12 + assert utc_value.tzinfo == timezone.utc + + +@pytest.mark.parametrize( + "raw,expected_hour", + [ + ("2026-07-15T12:00:00Z", 12), + ("2026-07-15T12:00:00+00:00", 12), + ("2026-07-15 12:00:00", 12), + ("2026-07-15T08:00:00-04:00", 12), + ], +) +def test_parse_to_utc_handles_common_formats(raw, expected_hour): + parsed = parse_to_utc(raw) + assert parsed is not None + assert parsed.tzinfo == timezone.utc + assert parsed.hour == expected_hour + + +def test_to_utc_iso_includes_offset(): + iso = to_utc_iso("2026-07-15T12:00:00") + assert iso.endswith("+00:00") or iso.endswith("Z") + assert "2026-07-15T12:00:00" in iso.replace("+00:00", "").replace("Z", "") + + +def test_to_utc_iso_current_time_has_offset(): + iso = to_utc_iso() + parsed = parse_to_utc(iso) + assert parsed is not None + assert parsed.tzinfo == timezone.utc + + +def test_format_utc_display_uses_utc_label(): + assert format_utc_display("2026-07-15T12:30:00Z") == "Jul 15, 2026 12:30 UTC" + assert format_utc_display("") == "Unknown" + assert format_utc_display(None) == "Unknown" + + +def test_report_payload_generated_at_is_utc_iso(): + task = { + "id": "task-1", + "tool_name": "nmap", + "target": "example.com", + "status": "completed", + "created_at": "2026-07-15 10:00:00", + "preset": "quick", + "command_used": "nmap example.com", + "inputs": {}, + } + result = { + "findings": [ + { + "title": "Open port", + "severity": "info", + "discovered_at": "2026-07-15T09:00:00", + } + ], + "structured": {}, + "summary": [], + "errors": [], + } + payload = ReportGenerator._build_report_payload(task, result) + + generated = parse_to_utc(payload["generated_at"]) + assert generated is not None + assert generated.tzinfo == timezone.utc + assert "+" in payload["generated_at"] or payload["generated_at"].endswith("Z") + + created = parse_to_utc(payload["created_at"]) + assert created is not None + assert created.hour == 10 + + finding = payload["findings"][0] + discovered = parse_to_utc(finding["discovered_at"]) + assert discovered is not None + assert discovered.hour == 9 + assert finding["discovered_at"].endswith("+00:00") or finding["discovered_at"].endswith("Z") + + +def test_report_format_timestamp_is_utc_display(): + assert ReportGenerator._format_timestamp("2026-07-15T12:30:00Z") == "Jul 15, 2026 12:30 UTC" + assert ReportGenerator._format_timestamp("2026-07-15 12:30:00") == "Jul 15, 2026 12:30 UTC" + + +def test_parse_discovered_at_returns_aware_utc(): + naive = _parse_discovered_at({"discovered_at": "2026-07-15T12:00:00"}) + assert naive.tzinfo == timezone.utc + + offset = _parse_discovered_at({"discovered_at": "2026-07-15T08:00:00-04:00"}) + assert offset.tzinfo == timezone.utc + assert offset.hour == 12 + + missing = _parse_discovered_at({}) + assert missing.tzinfo == timezone.utc + + +def test_deserialize_finding_rows_normalizes_discovered_at(): + rows = [ + { + "id": "f1", + "discovered_at": "2026-07-15 12:00:00", + "first_seen_at": "2026-07-14T12:00:00Z", + "last_seen_at": "2026-07-15T12:00:00+00:00", + "metadata_json": "{}", + } + ] + findings = deserialize_finding_rows(rows) + assert findings[0]["discovered_at"].endswith("+00:00") or findings[0]["discovered_at"].endswith("Z") + assert findings[0]["first_seen_at"].endswith("+00:00") or findings[0]["first_seen_at"].endswith("Z") + assert parse_to_utc(findings[0]["discovered_at"]).hour == 12 From 683843a966ddc0db2aa460b70651352f8ce565ae Mon Sep 17 00:00:00 2001 From: Naitik Verma Date: Wed, 15 Jul 2026 17:51:38 +0000 Subject: [PATCH 2/2] fix(backend): preserve UTC micros and update TLS stream mocks Default to_utc_iso to timespec=auto so finding intelligence tests can compare against datetime.now(UTC). Update TLS verification mocks for crawler client.stream() and stub crawl_target in API scanner tests. --- backend/secuscan/time_utils.py | 5 +- testing/backend/unit/test_tls_verification.py | 108 +++++++++++------- 2 files changed, 69 insertions(+), 44 deletions(-) diff --git a/backend/secuscan/time_utils.py b/backend/secuscan/time_utils.py index 63fbe7fff..19efff7ba 100644 --- a/backend/secuscan/time_utils.py +++ b/backend/secuscan/time_utils.py @@ -63,11 +63,14 @@ def parse_to_utc(value: Any) -> Optional[datetime]: return ensure_utc(parsed) -def to_utc_iso(value: Any = None, *, timespec: str = "seconds") -> str: +def to_utc_iso(value: Any = None, *, timespec: str = "auto") -> str: """Convert a value to an ISO-8601 UTC string with an explicit offset. ``None`` (or omitted) uses the current UTC time. Unparseable values fall back to ``utc_now()`` so API responses always include a timezone offset. + + ``timespec="auto"`` preserves microseconds when present so callers that + compare against ``datetime.now(timezone.utc)`` stay consistent. """ parsed = utc_now() if value is None else parse_to_utc(value) if parsed is None: diff --git a/testing/backend/unit/test_tls_verification.py b/testing/backend/unit/test_tls_verification.py index be8281d83..49ca33fae 100644 --- a/testing/backend/unit/test_tls_verification.py +++ b/testing/backend/unit/test_tls_verification.py @@ -19,6 +19,42 @@ import pytest +def _stream_response(mock_response: MagicMock) -> MagicMock: + """Build an async context manager for ``client.stream(...)`` (crawler uses streaming).""" + body = getattr(mock_response, "text", "") or "" + if isinstance(body, str): + body_bytes = body.encode("utf-8") + else: + body_bytes = body + + async def aiter_bytes(): + yield body_bytes + + mock_response.aiter_bytes = aiter_bytes + stream_cm = MagicMock() + stream_cm.__aenter__ = AsyncMock(return_value=mock_response) + stream_cm.__aexit__ = AsyncMock(return_value=False) + return stream_cm + + +def _mock_async_client(mock_response: MagicMock | None = None) -> AsyncMock: + """AsyncClient stand-in that supports both ``get`` and ``stream``.""" + if mock_response is None: + mock_response = MagicMock() + mock_response.text = "" + mock_response.url = "https://example.com/" + mock_response.status_code = 200 + mock_response.headers = {} + mock_response.history = [] + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.stream = MagicMock(return_value=_stream_response(mock_response)) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + return mock_client + + # --------------------------------------------------------------------------- # Settings defaults # --------------------------------------------------------------------------- @@ -54,17 +90,7 @@ def test_env_override_true(self, monkeypatch): class TestCrawlerVerifySsl: @pytest.mark.asyncio async def test_crawl_target_passes_verify_ssl(self): - mock_response = MagicMock() - mock_response.text = "" - mock_response.url = "https://example.com/" - mock_response.status_code = 200 - mock_response.headers = {} - mock_response.history = [] - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client.__aenter__ = AsyncMock(return_value=mock_client) - mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client = _mock_async_client() with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=mock_client) as mock_cls: with patch("backend.secuscan.crawler.settings") as mock_settings: @@ -79,17 +105,7 @@ async def test_crawl_target_passes_verify_ssl(self): @pytest.mark.asyncio async def test_crawl_target_verify_ssl_false_when_disabled(self): - mock_response = MagicMock() - mock_response.text = "" - mock_response.url = "https://example.com/" - mock_response.status_code = 200 - mock_response.headers = {} - mock_response.history = [] - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client.__aenter__ = AsyncMock(return_value=mock_client) - mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client = _mock_async_client() with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=mock_client) as mock_cls: with patch("backend.secuscan.crawler.settings") as mock_settings: @@ -114,21 +130,24 @@ async def test_api_scanner_passes_verify_ssl(self): scanner = APIScanner("task-verify", None) - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=MagicMock(status_code=404, text="", headers={})) - mock_client.__aenter__ = AsyncMock(return_value=mock_client) - mock_client.__aexit__ = AsyncMock(return_value=False) + mock_response = MagicMock(status_code=404, text="", headers={}) + mock_client = _mock_async_client(mock_response) + empty_crawl = {"api_hints": [], "links": [], "forms": [], "scripts": [], "params": []} with patch("backend.secuscan.scanners.api_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: with patch("backend.secuscan.scanners.api_scanner.settings") as mock_settings: mock_settings.verify_ssl = True - with patch.object(scanner, "_fetch_spec", return_value=None): - with patch.object(scanner, "_probe_graphql", return_value=([], [])): - await scanner.run("https://example.com", {}) - - mock_cls.assert_called() - _, kwargs = mock_cls.call_args - assert kwargs["verify"] is True + with patch( + "backend.secuscan.scanners.api_scanner.crawl_target", + new=AsyncMock(return_value=empty_crawl), + ): + with patch.object(scanner, "_fetch_spec", return_value=None): + with patch.object(scanner, "_probe_graphql", return_value=([], [])): + await scanner.run("https://example.com", {}) + + mock_cls.assert_called() + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is True @pytest.mark.asyncio async def test_api_scanner_verify_ssl_false_when_disabled(self): @@ -136,20 +155,23 @@ async def test_api_scanner_verify_ssl_false_when_disabled(self): scanner = APIScanner("task-no-verify", None) - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=MagicMock(status_code=404, text="", headers={})) - mock_client.__aenter__ = AsyncMock(return_value=mock_client) - mock_client.__aexit__ = AsyncMock(return_value=False) + mock_response = MagicMock(status_code=404, text="", headers={}) + mock_client = _mock_async_client(mock_response) + empty_crawl = {"api_hints": [], "links": [], "forms": [], "scripts": [], "params": []} with patch("backend.secuscan.scanners.api_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls: with patch("backend.secuscan.scanners.api_scanner.settings") as mock_settings: mock_settings.verify_ssl = False - with patch.object(scanner, "_fetch_spec", return_value=None): - with patch.object(scanner, "_probe_graphql", return_value=([], [])): - await scanner.run("https://example.com", {}) - - _, kwargs = mock_cls.call_args - assert kwargs["verify"] is False + with patch( + "backend.secuscan.scanners.api_scanner.crawl_target", + new=AsyncMock(return_value=empty_crawl), + ): + with patch.object(scanner, "_fetch_spec", return_value=None): + with patch.object(scanner, "_probe_graphql", return_value=([], [])): + await scanner.run("https://example.com", {}) + + _, kwargs = mock_cls.call_args + assert kwargs["verify"] is False # ---------------------------------------------------------------------------