fix(backend): standardize timezone handling to UTC ISO-8601

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,18 +1534,20 @@ 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}",
                     owner_id,
                     task_id,
                     f"{plugin.name} Report",
                     "technical",
+                    to_utc_iso(),
                     "ready" if status == TaskStatus.COMPLETED.value else "failed",
                     len(findings_data),
                     1,
@@ -1606,18 +1603,20 @@ 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}",
                     owner_id,
                     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

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:

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:

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"),

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):

backend/secuscan/time_utils.py

@@ -0,0 +1,86 @@
+"""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 = "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:
+        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)

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()
     })