[object Object]

← back to Unclaimed Property Platform

Cycle 4: runnable masked-search service + REAL rate limiter (closes Sec 1.3)

bdd1a4b9ddf0b900ae6653727289c60bc8184dce · 2026-08-01 19:31:42 -0700 · Steve Abrams

- services/common/rate_limit.py: SlidingWindowRateLimiter (injectable clock, per-key budgets,
  retry_after). Replaces the require_rate_limit_STUB masquerade — anti-enumeration is now real.
- services/search/service.py: SearchService composes validate -> rate-limit -> masked_search
  -> assert_public_safe (fail-closed) so every transport enforces identical privacy controls.
- services/search/run_local.py: dependency-free stdlib HTTP endpoint ($0). SMOKE-TESTED over
  real HTTP: 200 masked JSON, 400 empty-query, 429 after budget, /healthz. Single-threaded
  (SQLite is thread-bound; prod = Postgres pool).
- tests/test_cycle5_search.py: deterministic (injected clock) proof of masked results, real
  429 + window refill, empty-query rejection, per-client-key isolation.
Tests: 5/5 suites green. All local/synthetic/$0.

TK-10097

Files touched

Diff

commit bdd1a4b9ddf0b900ae6653727289c60bc8184dce
Author: Steve Abrams <steve@designerwallcoverings.com>
Date:   Sat Aug 1 19:31:42 2026 -0700

    Cycle 4: runnable masked-search service + REAL rate limiter (closes Sec 1.3)
    
    - services/common/rate_limit.py: SlidingWindowRateLimiter (injectable clock, per-key budgets,
      retry_after). Replaces the require_rate_limit_STUB masquerade — anti-enumeration is now real.
    - services/search/service.py: SearchService composes validate -> rate-limit -> masked_search
      -> assert_public_safe (fail-closed) so every transport enforces identical privacy controls.
    - services/search/run_local.py: dependency-free stdlib HTTP endpoint ($0). SMOKE-TESTED over
      real HTTP: 200 masked JSON, 400 empty-query, 429 after budget, /healthz. Single-threaded
      (SQLite is thread-bound; prod = Postgres pool).
    - tests/test_cycle5_search.py: deterministic (injected clock) proof of masked results, real
      429 + window refill, empty-query rejection, per-client-key isolation.
    Tests: 5/5 suites green. All local/synthetic/$0.
    
    TK-10097
---
 services/search/run_local.py |  98 ++++++++++++++++++++++++++++++++++++++++
 tests/test_cycle5_search.py  | 105 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 203 insertions(+)

diff --git a/services/search/run_local.py b/services/search/run_local.py
new file mode 100644
index 0000000..b6328f7
--- /dev/null
+++ b/services/search/run_local.py
@@ -0,0 +1,98 @@
+"""Runnable anonymous-search HTTP endpoint on the Python stdlib (no third-party deps, $0).
+
+This is the dependency-free transport over SearchService — it lets the whole anti-enumeration
++ masking path run and be smoke-tested without FastAPI/uvicorn/OpenSearch. Production uses
+the FastAPI reference in search_api.py; this proves the core end-to-end over real HTTP.
+
+    python -m services.search.run_local            # seeds synthetic data, serves :8799
+    curl 'http://127.0.0.1:8799/v1/search?q=Oneil'
+    curl 'http://127.0.0.1:8799/healthz'
+
+Env: UPP_PORT (default 8799), UPP_RATE_MAX (default 30), UPP_RATE_WINDOW (default 60).
+Client-key = the direct peer IP (trusted for a local direct-connect server). Behind a real
+proxy, derive the key from the trusted proxy hop, never a raw X-Forwarded-For header.
+"""
+from __future__ import annotations
+
+import json
+import os
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from urllib.parse import parse_qs, urlparse
+
+from services.common.rate_limit import SlidingWindowRateLimiter
+from services.common.sqlite_repo import FileObjectStore, SqliteRepository
+from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed
+from services.search.service import EmptyQuery, RateLimited, SearchService
+
+_SAMPLE = Path(__file__).resolve().parents[2] / "data" / "sample" / "sample_state_feed.csv"
+
+
+def build_service() -> SearchService:
+    import tempfile
+    tmp = Path(tempfile.mkdtemp(prefix="upp-serve-"))
+    store = FileObjectStore(tmp / "obj")
+    (tmp / "obj").mkdir(parents=True, exist_ok=True)
+    store.write_bytes("incoming/s.csv", _SAMPLE.read_bytes())
+    repo = SqliteRepository(str(tmp / "serve.db"))
+    ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/s.csv"), store, repo)
+    limiter = SlidingWindowRateLimiter(
+        max_requests=int(os.environ.get("UPP_RATE_MAX", "30")),
+        window_seconds=float(os.environ.get("UPP_RATE_WINDOW", "60")),
+    )
+    return SearchService(repository=repo, limiter=limiter)
+
+
+def make_handler(service: SearchService):
+    class Handler(BaseHTTPRequestHandler):
+        def _json(self, code: int, body: dict, headers: dict | None = None) -> None:
+            payload = json.dumps(body).encode()
+            self.send_response(code)
+            self.send_header("Content-Type", "application/json")
+            for k, v in (headers or {}).items():
+                self.send_header(k, str(v))
+            self.end_headers()
+            self.wfile.write(payload)
+
+        def do_GET(self) -> None:  # noqa: N802
+            parsed = urlparse(self.path)
+            if parsed.path == "/healthz":
+                self._json(200, {"status": "ok"})
+                return
+            if parsed.path != "/v1/search":
+                self._json(404, {"error": "not found"})
+                return
+            qs = parse_qs(parsed.query)
+            query = (qs.get("q", [""])[0])
+            client_key = f"ip:{self.client_address[0]}"
+            try:
+                results = service.search(query, client_key=client_key)
+                self._json(200, {"results": results, "count": len(results)})
+            except EmptyQuery as e:
+                self._json(400, {"error": str(e)})
+            except RateLimited as e:
+                self._json(429, {"error": "rate limited"},
+                           headers={"Retry-After": int(e.retry_after) + 1})
+
+        def log_message(self, *args) -> None:  # quiet
+            pass
+
+    return Handler
+
+
+def main() -> int:
+    port = int(os.environ.get("UPP_PORT", "8799"))
+    service = build_service()
+    # Single-threaded: the SQLite connection is bound to this thread. Requests serialize,
+    # which is fine for a local demo. Production (Postgres + connection pool) is thread-safe.
+    httpd = HTTPServer(("127.0.0.1", port), make_handler(service))
+    print(f"serving synthetic unclaimed-property search on http://127.0.0.1:{port}")
+    try:
+        httpd.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_cycle5_search.py b/tests/test_cycle5_search.py
new file mode 100644
index 0000000..e2a98c8
--- /dev/null
+++ b/tests/test_cycle5_search.py
@@ -0,0 +1,105 @@
+"""Cycle 4 tests — SearchService: masked results + REAL rate limiting + fail-closed projection.
+
+Run:  python -m tests.test_cycle5_search
+
+Proves (deterministic, no sleeps — the limiter clock is injected):
+  1. search() returns masked, projection-safe rows (no raw PII).
+  2. rate limiting is REAL — N allowed per window, N+1 raises RateLimited; budget refills
+     after the window passes. (Closes Security 1.3, the last open High.)
+  3. empty/punctuation queries raise EmptyQuery.
+  4. rate-limit budgets are per-client-key (one abuser can't exhaust another's budget).
+"""
+from __future__ import annotations
+
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from services.common.rate_limit import SlidingWindowRateLimiter
+from services.common.sqlite_repo import FileObjectStore, SqliteRepository
+from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed
+from services.search.service import EmptyQuery, RateLimited, SearchService
+
+SAMPLE = Path(__file__).resolve().parents[1] / "data" / "sample" / "sample_state_feed.csv"
+
+
+def ok(msg: str) -> None:
+    print(f"  ✓ {msg}")
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.t = 0.0
+
+    def __call__(self) -> float:
+        return self.t
+
+
+def _populated_repo(tmp: Path) -> SqliteRepository:
+    store = FileObjectStore(tmp / "obj")
+    (tmp / "obj").mkdir(parents=True, exist_ok=True)
+    store.write_bytes("incoming/s.csv", SAMPLE.read_bytes())
+    repo = SqliteRepository(str(tmp / "p.db"))
+    ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/s.csv"), store, repo)
+    return repo
+
+
+def main() -> int:
+    tmp = Path(tempfile.mkdtemp(prefix="upp-cycle5-"))
+    repo = _populated_repo(tmp)
+    clock = FakeClock()
+    limiter = SlidingWindowRateLimiter(max_requests=3, window_seconds=60, clock=clock)
+    svc = SearchService(repository=repo, limiter=limiter, max_results=20)
+
+    print("1) masked, projection-safe results")
+    rows = svc.search("Oneil", client_key="ip:10.0.0.1")
+    assert rows, "expected a hit"
+    assert all("owner_name_masked" in r and "owner_name_raw" not in r for r in rows), rows
+    assert "ONEIL" not in (rows[0]["owner_name_masked"] or "").upper(), rows[0]
+    ok(f"hit {rows[0]['owner_name_masked']} | {rows[0]['amount_band']} (no raw PII)")
+
+    print("2) REAL rate limiting (3 per 60s window, deterministic clock)")
+    limiter.reset()
+    k = "ip:1.2.3.4"
+    for i in range(3):
+        svc.search("Oneil", client_key=k)          # 3 allowed at t=0
+    raised = False
+    try:
+        svc.search("Oneil", client_key=k)          # 4th within window -> blocked
+    except RateLimited as e:
+        raised = True
+        assert e.retry_after > 0, e.retry_after
+    assert raised, "4th request in window should have been rate-limited"
+    ok("4th request in window -> RateLimited (retry_after > 0)")
+
+    clock.t = 61.0                                   # window elapsed
+    rows_after = svc.search("Oneil", client_key=k)   # budget refilled
+    assert rows_after, "should be allowed again after the window"
+    ok("after window elapses -> allowed again")
+
+    print("3) empty / punctuation queries rejected")
+    for bad in ("", "   ", "!!!"):
+        limiter.reset()
+        try:
+            svc.search(bad, client_key="ip:fresh")
+            raise AssertionError(f"empty query {bad!r} should raise EmptyQuery")
+        except EmptyQuery:
+            pass
+    ok("empty/punctuation queries -> EmptyQuery")
+
+    print("4) per-client-key budgets are independent")
+    limiter.reset()
+    for _ in range(3):
+        svc.search("Oneil", client_key="ip:aaa")     # exhaust aaa
+    # a DIFFERENT client is unaffected
+    assert svc.search("Oneil", client_key="ip:bbb"), "bbb should have its own budget"
+    ok("one client's exhausted budget does not block another")
+
+    print("\nALL CYCLE-4 SEARCH-SERVICE ASSERTIONS PASSED ✅")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

← 3d2a81a auto-save: 2026-08-01T11:36:26 (2 files) — services/common/r  ·  back to Unclaimed Property Platform  ·  docs: ledger Cycle-4 record (search service + real limiter); 9a96426 →