`services/search/search_api.py`

112 lines
"""Masked public search API (FastAPI + OpenSearch reference).

This is the PRODUCTION reference. It requires `fastapi` and `opensearch-py`
(see requirements.txt) and a running OpenSearch. For the dependency-free prototype
smoke test, the in-DB masked_search() on SqliteRepository is used instead.

Design invariants:
  - anonymous search is a SEPARATE service from claims; the authoritative property table
    is never internet-queryable.
  - only jurisdiction-approved, MASKED projections are returned (masked name, city,
    holder, type, amount BAND, state reference) — never SSNs, full addresses, exact
    amounts, or claim evidence.
  - fuzzy queries are constrained (filters, prefix_length, max_expansions, result caps)
    because broad fuzzy/wildcard queries are resource-expensive and enable enumeration.
"""
from __future__ import annotations

import os
from typing import Annotated, Any

try:
    from fastapi import Depends, FastAPI, Header, HTTPException, Query
    from opensearchpy import OpenSearch
    from pydantic import BaseModel
    _DEPS = True
except ImportError:  # keep the module importable for docs/tests without the deps
    _DEPS = False
    BaseModel = object  # type: ignore


if _DEPS:
    app = FastAPI(title="National Unclaimed Property Search API")

    search_client = OpenSearch(
        hosts=[os.environ.get("OPENSEARCH_URL", "https://localhost:9200")],
        http_auth=(os.environ.get("OPENSEARCH_USERNAME", ""),
                   os.environ.get("OPENSEARCH_PASSWORD", "")),
        use_ssl=True, verify_certs=True,
    )

    class SearchResult(BaseModel):
        public_reference: str
        jurisdiction: str
        owner_name_masked: str
        owner_city: str | None
        holder_name: str
        property_type: str | None
        amount_band: str | None

    class SearchResponse(BaseModel):
        results: list[SearchResult]
        total_relation: str
        next_page_token: str | None = None

    def require_rate_limit_STUB(
        forwarded_for: Annotated[str | None, Header()] = None,
    ) -> None:
        """SECURITY: THIS IS NOT A RATE LIMITER. It is a placeholder.

        Anti-enumeration REQUIRES a real limiter keyed off the TRUSTED-proxy IP (never the
        spoofable X-Forwarded-For header) plus per-session + global-velocity limits and
        anomaly detection. Until that exists, this endpoint MUST NOT be exposed to the
        internet with real data. The name ends in _STUB so no reviewer mistakes it for a
        control. (Security finding 1.3.)"""
        if forwarded_for and len(forwarded_for) > 500:
            raise HTTPException(status_code=400, detail="Invalid forwarding header")

    @app.get("/v1/search", response_model=SearchResponse)
    def search_properties(
        last_or_business: Annotated[str, Query(min_length=2, max_length=100)],
        first_name: Annotated[str | None, Query(max_length=100)] = None,
        jurisdiction: Annotated[str | None, Query(min_length=2, max_length=3)] = None,
        city: Annotated[str | None, Query(max_length=100)] = None,
        limit: Annotated[int, Query(ge=1, le=50)] = 20,
        _: None = Depends(require_rate_limit_STUB),
    ) -> "SearchResponse":
        must: list[dict[str, Any]] = [{
            "multi_match": {
                "query": last_or_business,
                "fields": ["owner_name_normalized^5", "owner_name_tokens^3",
                           "business_name_normalized^4"],
                "type": "best_fields",
                "fuzziness": "AUTO:4,7", "prefix_length": 1, "max_expansions": 25,
            }
        }]
        filters: list[dict[str, Any]] = [
            {"term": {"is_public": True}},
            {"term": {"is_suppressed": False}},
        ]
        if first_name:
            must.append({"match": {"first_name_normalized": {
                "query": first_name, "fuzziness": "AUTO"}}})
        if jurisdiction:
            filters.append({"term": {"jurisdiction": jurisdiction.upper()}})
        if city:
            filters.append({"match": {"city_normalized": city}})

        response = search_client.search(
            index="unclaimed-property-public",
            body={
                "size": limit, "track_total_hits": False,
                "_source": ["public_reference", "jurisdiction", "owner_name_masked",
                            "owner_city", "holder_name", "property_type", "amount_band"],
                "query": {"bool": {"must": must, "filter": filters}},
            },
        )
        hits = response["hits"]["hits"]
        return SearchResponse(
            results=[SearchResult(**h["_source"]) for h in hits],
            total_relation=response["hits"]["total"]["relation"],
        )