Merge pull request #162 from abinaze/fix/webhook-ssrf-and-key-rotation

backend/services/api_key_manager.py

@@ -9,6 +9,7 @@ Roles:
   admin   - all methods including key management
 """
 import json
+import os
 import uuid
 import hashlib
 import secrets
@@ -74,6 +75,43 @@ def _save_key(entry: Dict[str, Any]) -> None:
         logger.error(f"Failed to save API key: {e}")
 
 
+def _rewrite_keys_file(keys: Dict[str, Dict[str, Any]]) -> None:
+    """Atomically rewrite the whole keys file with exactly one line per
+    key (F-10).
+
+    Replaces the previous append-only-forever behavior: _save_key() used
+    to append a full copy of an entry on every successful authentication
+    (not just on creation), so data/api_keys.jsonl grew by one line per
+    auth with no bound, and _load_keys() re-read and re-parsed every
+    accumulated line on every single call. Writes to a temp file first
+    and os.replace()'s it into place, which is atomic on both POSIX and
+    Windows -- a crash mid-write can never leave a corrupt/partial file
+    in KEYS_PATH.
+    """
+    KEYS_PATH.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = KEYS_PATH.with_suffix(KEYS_PATH.suffix + ".tmp")
+    with open(tmp_path, "w", encoding="utf-8") as f:
+        for entry in keys.values():
+            f.write(json.dumps(entry) + "\n")
+    os.replace(tmp_path, KEYS_PATH)
+
+
+def _upsert_key(entry: Dict[str, Any]) -> None:
+    """Insert or update a single key entry, compacted (F-10).
+
+    Performs the full load-merge-rewrite sequence under one lock
+    acquisition, so concurrent callers can no longer race on a stale
+    read -- this is also what closes the use_count undercount the audit
+    flagged alongside the plain unbounded-growth issue (previously only
+    the final _save_key() append was lock-protected, not the read and
+    increment before it).
+    """
+    with _key_write_lock:
+        keys = _load_keys()
+        keys[entry["key_id"]] = entry
+        _rewrite_keys_file(keys)
+
+
 def create_key(name: str, role: str = "analyst",
                description: str = "", created_by: str = "system") -> Dict[str, Any]:
     if role not in ROLES:
@@ -92,7 +130,7 @@ def create_key(name: str, role: str = "analyst",
         "key_hash": key_hash, "salt": salt, "created_at": _now(), "created_by": created_by,
         "last_used": None, "use_count": 0, "active": True,
     }
-    _save_key(entry)
+    _upsert_key(entry)
     logger.info(f"API key created: {key_id} name={name} role={role}")
     return {**entry, "key": raw_key,
             "warning": "Save this key now. It will not be shown again.",
@@ -103,7 +141,7 @@ def verify_key(raw_key: str) -> Optional[Dict[str, Any]]:
     if not raw_key or not raw_key.startswith("vfx_"):
         return None
     keys = _load_keys()
-    for entry in keys.values():
+    for key_id, entry in keys.items():
         # salt defaults to "" for legacy records created before F-6, which
         # reproduces the exact old unsalted hash -- so already-issued keys
         # keep verifying correctly with no migration step required.
@@ -112,24 +150,34 @@ def verify_key(raw_key: str) -> Optional[Dict[str, Any]]:
             secrets.compare_digest(entry.get("key_hash", ""), expected_hash)
             and entry.get("active", False)
         ):
-            entry["last_used"] = _now()
-            entry["use_count"] = entry.get("use_count", 0) + 1
+            # F-10: re-read the latest on-disk state and increment
+            # use_count inside the SAME lock as the write. Previously
+            # only the final _save_key() append was lock-protected, not
+            # the read-and-increment before it -- two concurrent
+            # requests using the same key could both read the same
+            # stale use_count, and one increment would be lost.
             with _key_write_lock:
-                _save_key(entry)
-            return {k: v for k, v in entry.items() if k not in ("key_hash", "salt")}
+                latest_keys = _load_keys()
+                latest_entry = latest_keys.get(key_id, entry)
+                latest_entry["last_used"] = _now()
+                latest_entry["use_count"] = latest_entry.get("use_count", 0) + 1
+                latest_keys[key_id] = latest_entry
+                _rewrite_keys_file(latest_keys)
+            return {k: v for k, v in latest_entry.items() if k not in ("key_hash", "salt")}
     return None
 
 
 
 def revoke_key(key_id: str, revoked_by: str = "system") -> Dict[str, Any]:
-    keys = _load_keys()
-    if key_id not in keys:
-        return {"error": f"Key not found: {key_id}"}
-    entry = keys[key_id]
-    entry["active"]     = False
-    entry["revoked_at"] = _now()
-    entry["revoked_by"] = revoked_by
-    _save_key(entry)
+    with _key_write_lock:
+        keys = _load_keys()
+        if key_id not in keys:
+            return {"error": f"Key not found: {key_id}"}
+        entry = keys[key_id]
+        entry["active"]     = False
+        entry["revoked_at"] = _now()
+        entry["revoked_by"] = revoked_by
+        _rewrite_keys_file(keys)
     logger.info(f"API key revoked: {key_id}")
     return {k: v for k, v in entry.items() if k not in ("key_hash", "salt")}
 

backend/services/webhook_manager.py

@@ -340,6 +340,25 @@ def _attempt_delivery(
     delivery_id: str,
 ) -> tuple[bool, Optional[int], Optional[str]]:
     """Perform a single HTTP POST. Returns (success, status_code, error)."""
+    # F-9: re-run the same SSRF guard used at registration, immediately
+    # before every delivery attempt (including each retry). The
+    # registration-time check alone leaves a DNS-rebinding window open --
+    # a hostname can resolve to a safe address at registration and to an
+    # internal/metadata address later, especially across the retry delays
+    # in _deliver_with_retry (up to _RETRY_DELAYS apart). Treated as a
+    # delivery failure (not a crash) so it flows through the existing
+    # retry/suspend/logging machinery, but logged distinctly so a real
+    # rebinding attempt is visible and not indistinguishable from an
+    # ordinary network failure.
+    try:
+        _reject_unsafe_webhook_target(url)
+    except ValueError as exc:
+        logger.error(
+            "Webhook %s delivery blocked at send-time SSRF check: %s",
+            webhook_id, exc,
+        )
+        return False, None, f"Blocked by delivery-time SSRF guard: {exc}"
+
     req = _URLRequest(
         url,
         data=body,

backend/tests/test_api_keys.py

@@ -149,3 +149,81 @@ def test_api_verify_endpoint_invalid(client):
     response = client.get("/api/v1/keys/verify",
                           headers={"Authorization": "Bearer vfx_invalid"})
     assert response.status_code == 401
+
+
+# ── F-10: compacted storage (no unbounded growth) + use_count race ─────────
+
+def test_repeated_auth_does_not_grow_the_file(temp_keys_file):
+    """F-10 regression test: the file must stay at exactly one line per
+    key regardless of how many times that key is used to authenticate --
+    previously _save_key() appended a full copy on every successful
+    verify_key() call, so a single key used 100 times produced 100+
+    lines instead of 1."""
+    from backend.services.api_key_manager import create_key, verify_key
+
+    created = create_key("Repeated Use Key", role="analyst")
+    for _ in range(25):
+        result = verify_key(created["key"])
+        assert result is not None
+
+    line_count = sum(1 for line in temp_keys_file.read_text().splitlines() if line.strip())
+    assert line_count == 1, (
+        f"expected exactly 1 line for 1 key after 25 authentications, got {line_count} -- "
+        f"the file is growing unbounded again (F-10)"
+    )
+
+
+def test_multiple_keys_stay_one_line_each(temp_keys_file):
+    from backend.services.api_key_manager import create_key, verify_key
+
+    keys = [create_key(f"Key {i}", role="analyst") for i in range(5)]
+    for k in keys:
+        verify_key(k["key"])
+        verify_key(k["key"])
+
+    line_count = sum(1 for line in temp_keys_file.read_text().splitlines() if line.strip())
+    assert line_count == 5, f"expected exactly 5 lines for 5 keys, got {line_count}"
+
+
+def test_use_count_accurate_under_concurrent_verification():
+    """F-10 regression test: concurrent requests using the SAME key must
+    not lose increments. Previously only the final _save_key() append
+    was lock-protected, not the read-and-increment before it, so two
+    threads could read the same stale use_count and one increment would
+    be silently lost.
+    """
+    import threading
+    from backend.services.api_key_manager import create_key, verify_key, list_keys
+
+    created = create_key("Concurrency Test Key", role="analyst")
+    raw_key = created["key"]
+
+    N_THREADS = 20
+    barrier = threading.Barrier(N_THREADS)
+
+    def _do_verify():
+        barrier.wait()  # maximize actual overlap
+        verify_key(raw_key)
+
+    threads = [threading.Thread(target=_do_verify) for _ in range(N_THREADS)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=5)
+
+    entries = list_keys(include_inactive=True)
+    entry = next(e for e in entries if e["key_id"] == created["key_id"])
+    assert entry["use_count"] == N_THREADS, (
+        f"expected use_count == {N_THREADS} after {N_THREADS} concurrent verifications, "
+        f"got {entry['use_count']} -- increments were lost to the race (F-10)"
+    )
+
+
+def test_revoke_does_not_duplicate_lines(temp_keys_file):
+    from backend.services.api_key_manager import create_key, revoke_key
+
+    created = create_key("Revoke Compaction Test", role="analyst")
+    revoke_key(created["key_id"])
+
+    line_count = sum(1 for line in temp_keys_file.read_text().splitlines() if line.strip())
+    assert line_count == 1, f"expected exactly 1 line after create+revoke, got {line_count}"

backend/tests/test_webhooks.py

@@ -202,3 +202,62 @@ class TestAPIAuth:
         """GET /api/v1/webhooks/deliveries without auth → 401."""
         resp = client.get("/api/v1/webhooks/deliveries")
         assert resp.status_code == 401
+
+
+class TestDeliveryTimeSSRFGuard:
+    """F-9 regression tests: the SSRF guard must run again immediately
+    before each delivery attempt, not just once at registration -- a
+    hostname can resolve to a safe address at registration time and to
+    an internal/metadata address later (DNS rebinding), especially
+    across the retry delays in _deliver_with_retry.
+    """
+
+    def test_attempt_delivery_blocks_when_target_becomes_unsafe(self, monkeypatch):
+        """Simulates DNS rebinding: the target was safe at registration,
+        but resolves to a disallowed address by the time delivery is
+        attempted. _attempt_delivery must reject it and must NOT reach
+        the network at all.
+        """
+        import backend.services.webhook_manager as wm
+
+        def _now_unsafe(url):
+            raise ValueError(f"Webhook URL resolves to a disallowed address (simulated rebinding to 169.254.169.254).")
+
+        monkeypatch.setattr(wm, "_reject_unsafe_webhook_target", _now_unsafe)
+
+        network_was_called = []
+        def _fail_if_called(*args, **kwargs):
+            network_was_called.append(True)
+            raise AssertionError("must not reach the network once the SSRF guard rejects the target")
+        monkeypatch.setattr(wm, "_URLRequest", _fail_if_called)
+
+        success, status_code, error = wm._attempt_delivery(
+            "https://example.com/hook", b"{}", "sig", "webhook-1", "delivery-1"
+        )
+
+        assert success is False
+        assert status_code is None
+        assert "SSRF" in error
+        assert network_was_called == [], "delivery attempted network access despite the SSRF guard rejecting it"
+
+    def test_attempt_delivery_proceeds_when_target_still_safe(self, monkeypatch):
+        """The common case: target is still safe at delivery time --
+        must not be blocked, and should reach the (mocked) network."""
+        import backend.services.webhook_manager as wm
+
+        monkeypatch.setattr(wm, "_reject_unsafe_webhook_target", lambda url: None)
+
+        class _FakeResponse:
+            status = 200
+            def __enter__(self): return self
+            def __exit__(self, *a): return False
+
+        monkeypatch.setattr(wm, "urlopen", lambda req, timeout=10: _FakeResponse())
+
+        success, status_code, error = wm._attempt_delivery(
+            "https://example.com/hook", b"{}", "sig", "webhook-1", "delivery-1"
+        )
+
+        assert success is True
+        assert status_code == 200
+        assert error is None