Merge pull request #320 from spoo-me/feat/anonymous-metadata

feat: anonymous metadata, expand, and domain-intel endpoints

Author: Zingzy

config.py

@@ -401,6 +401,29 @@ class MetaTagsSettings(BaseSettings):
     fetch_user_agent: str = "spoo.me-og-validator/1.0 (+https://spoo.me)"
 
 
+class WebRiskSettings(BaseSettings):
+    """Google Web Risk lookups for the URL expander's safety verdict.
+
+    Env vars are prefixed ``WEB_RISK_`` so a generic ``API_KEY`` set
+    elsewhere in the deploy environment can't configure this. Without a
+    key the check silently doesn't run and the verdict is absent — the
+    tool never shows a fabricated one.
+    """
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        extra="ignore",
+        env_prefix="WEB_RISK_",
+    )
+
+    api_key: str = ""
+    timeout_seconds: float = Field(default=4.0, gt=0)
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.api_key)
+
+
 class WebhookSettings(BaseSettings):
     """Webhooks system — real-time event deliveries to subscriber URLs.
 
@@ -603,6 +626,7 @@ def _password_max_length_sane(cls, v: int) -> int:
     edge_cache: EdgeCacheSettings | None = None
     r2: R2StorageSettings | None = None
     meta_tags: MetaTagsSettings | None = None
+    web_risk: WebRiskSettings | None = None
     webhooks: WebhookSettings | None = None
 
     @model_validator(mode="after")
@@ -643,6 +667,8 @@ def _populate_sub_configs_and_secret(self) -> AppSettings:
             self.r2 = R2StorageSettings()
         if self.meta_tags is None:
             self.meta_tags = MetaTagsSettings()
+        if self.web_risk is None:
+            self.web_risk = WebRiskSettings()
         if self.webhooks is None:
             self.webhooks = WebhookSettings()
         if self.webhooks.enabled and not self.secret_key:

dependencies/__init__.py

@@ -62,6 +62,7 @@
     CredentialSvc,
     CustomDomainSvc,
     DeviceAuthSvc,
+    DomainIntelSvc,
     ExportSvc,
     FeatureFlagSvc,
     OAuthSvc,
@@ -70,6 +71,7 @@
     ProfilePictureSvc,
     ReportIntakeSvc,
     StatsSvc,
+    UrlExpandSvc,
     UrlSvc,
     UserRepo,
     VerificationSvc,
@@ -122,6 +124,7 @@
     "CurrentUser",
     "CustomDomainSvc",
     "DeviceAuthSvc",
+    "DomainIntelSvc",
     "ExportSvc",
     "FeatureFlagSvc",
     "GeoIP",
@@ -139,6 +142,7 @@
     "ReportIntakeSvc",
     "Settings",
     "StatsSvc",
+    "UrlExpandSvc",
     "UrlSvc",
     "UserRepo",
     "VerificationSvc",

dependencies/services.py

@@ -27,6 +27,7 @@
 from services.click.sinks import ClickEventSink
 from services.contact_service import ContactService
 from services.custom_domain_service import CustomDomainService
+from services.domain_intel_service import DomainIntelService
 from services.export.service import ExportService
 from services.feature_flag_service import FeatureFlagService
 from services.oauth_service import OAuthService
@@ -36,6 +37,7 @@
 from services.public_stats_service import PublicStatsService
 from services.report_intake_service import ReportIntakeService
 from services.stats_service import StatsService
+from services.url_expand_service import UrlExpandService
 from services.url_service import UrlService
 from services.webhooks.service import WebhookService
 
@@ -136,6 +138,14 @@ def get_public_preview_service(request: Request) -> PublicPreviewService:
     return request.app.state.public_preview_service
 
 
+def get_url_expand_service(request: Request) -> UrlExpandService:
+    return request.app.state.url_expand_service
+
+
+def get_domain_intel_service(request: Request) -> DomainIntelService:
+    return request.app.state.domain_intel_service
+
+
 def get_report_intake_service(request: Request) -> ReportIntakeService:
     return request.app.state.report_intake_service
 
@@ -169,5 +179,7 @@ def get_webhook_service(request: Request) -> WebhookService:
 FeatureFlagSvc = Annotated[FeatureFlagService, Depends(get_feature_flag_service)]
 CustomDomainSvc = Annotated[CustomDomainService, Depends(get_custom_domain_service)]
 PublicPreviewSvc = Annotated[PublicPreviewService, Depends(get_public_preview_service)]
+UrlExpandSvc = Annotated[UrlExpandService, Depends(get_url_expand_service)]
+DomainIntelSvc = Annotated[DomainIntelService, Depends(get_domain_intel_service)]
 ReportIntakeSvc = Annotated[ReportIntakeService, Depends(get_report_intake_service)]
 WebhookSvc = Annotated[WebhookService, Depends(get_webhook_service)]

dependencies/wiring.py

@@ -55,6 +55,7 @@
 from services.click.sinks import InlineSink, RedisStreamSink
 from services.contact_service import ContactService
 from services.custom_domain_service import CustomDomainService
+from services.domain_intel_service import DomainIntelService
 from services.edge_cache.og_writethrough import OgEdgeWritethrough
 from services.events.sinks import (
     InlineDomainEventSink,
@@ -76,6 +77,7 @@
 from services.stats_service import StatsService
 from services.tenant_resolver import CachedMongoTenantResolver
 from services.token_factory import TokenFactory
+from services.url_expand_service import UrlExpandService
 from services.url_service import UrlService
 from services.webhooks import (
     DeliveryExecutor,
@@ -329,6 +331,18 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         system_default_domain=settings.system_default_domain,
     )
     app.state.public_preview_service = PublicPreviewService(public_link_resolver)
+    app.state.url_expand_service = UrlExpandService(
+        blocked_url_repo,
+        MetaFetchCache(redis_client, prefix="url_expand"),
+        regex_timeout=settings.blocked_url_regex_timeout,
+        user_agent=settings.meta_tags.fetch_user_agent,
+        http_client=http_client,
+        web_risk_api_key=settings.web_risk.api_key,
+    )
+    app.state.domain_intel_service = DomainIntelService(
+        MetaFetchCache(redis_client, prefix="domain_intel", ttl_seconds=86_400),
+        http_client,
+    )
     app.state.public_stats_service = PublicStatsService(
         public_link_resolver,
         app.state.stats_service,

infrastructure/cache/meta_fetch_cache.py

@@ -25,14 +25,15 @@ def __init__(
         *,
         ttl_seconds: int = 3600,
         negative_ttl_seconds: int = 300,
+        prefix: str = "meta_fetch",
     ) -> None:
         self._redis = redis_client
         self._ttl = ttl_seconds
         self._negative_ttl = negative_ttl_seconds
+        self._prefix = prefix
 
-    @staticmethod
-    def _key(url: str) -> str:
-        return f"meta_fetch:{hashlib.sha256(url.encode()).hexdigest()}"
+    def _key(self, url: str) -> str:
+        return f"{self._prefix}:{hashlib.sha256(url.encode()).hexdigest()}"
 
     async def get(self, url: str) -> dict | None:
         if self._redis is None:

infrastructure/safe_fetch.py

@@ -78,7 +78,7 @@ def _is_public(ip: str) -> bool:
     return addr.is_global and not addr.is_multicast
 
 
-async def _resolve_public_ip(host: str) -> str:
+async def resolve_public_ip(host: str) -> str:
     """Resolve *host* and return one address, rejecting any private result."""
     # A literal IP address skips DNS entirely.
     try:
@@ -153,7 +153,7 @@ async def fetch_public(
         parsed = httpx.URL(url)
         if parsed.scheme != "https":
             raise FetchHardError("non-https URL")
-        ip = await _resolve_public_ip(parsed.host)
+        ip = await resolve_public_ip(parsed.host)
 
         # Pin the connection to the validated IP; keep name-based TLS via
         # sni_hostname and the Host header.
@@ -282,7 +282,7 @@ async def post_public(
         parsed = httpx.URL(url)
         if parsed.scheme != "https":
             return PostResult(None, "non-https URL", None)
-        ip = await _resolve_public_ip(parsed.host)
+        ip = await resolve_public_ip(parsed.host)
     except (FetchHardError, FetchTransientError, httpx.InvalidURL) as exc:
         # Transient DNS failures included: delivery outcomes are DATA for
         # the retry ladder, never exceptions that skip attempt recording.
@@ -328,4 +328,82 @@ async def validate_public_https_url(url: str) -> None:
     parsed = httpx.URL(url)
     if parsed.scheme != "https":
         raise FetchHardError("URL must be https")
-    await _resolve_public_ip(parsed.host)
+    await resolve_public_ip(parsed.host)
+
+
+@dataclass(frozen=True)
+class ChainHop:
+    url: str
+    status: int | None  # None: this hop never answered
+
+
+@dataclass(frozen=True)
+class ExpandedChain:
+    hops: list[ChainHop]
+    final_url: str
+    final_status: int | None
+    truncated: bool
+
+
+async def expand_public(
+    url: str,
+    *,
+    timeout: float = 5.0,
+    max_redirects: int = 10,
+    user_agent: str = DEFAULT_USER_AGENT,
+) -> ExpandedChain:
+    """Follow *url*'s redirect chain hop by hop and report every stop.
+
+    Unlike fetch_public this never reads bodies and allows plain-http
+    hops — real shortener chains bounce through http trackers, and only
+    headers ever ride the wire here. Every hop still gets the full SSRF
+    guard: public-DNS resolution, IP pinning, no auto-redirects.
+    """
+    hops: list[ChainHop] = []
+    for _hop in range(max_redirects + 1):
+        parsed = httpx.URL(url)
+        if parsed.scheme not in ("http", "https"):
+            raise FetchHardError("unsupported scheme")
+        try:
+            ip = await resolve_public_ip(parsed.host)
+        except FetchHardError:
+            if not hops:
+                raise
+            # Mid-chain dead end (NXDOMAIN, private space): report the
+            # chain up to it rather than erasing what we learned.
+            hops.append(ChainHop(url, None))
+            return ExpandedChain(hops, url, None, False)
+        pinned = parsed.copy_with(host=_bracket(ip))
+        async with httpx.AsyncClient(follow_redirects=False, timeout=timeout) as client:
+            request = client.build_request(
+                "GET",
+                pinned,
+                headers={
+                    "Host": parsed.host,
+                    "User-Agent": user_agent,
+                    "Accept-Encoding": "identity",
+                },
+                extensions=(
+                    {"sni_hostname": parsed.host} if parsed.scheme == "https" else {}
+                ),
+            )
+            try:
+                resp = await client.send(request, stream=True)
+            except (httpx.TimeoutException, httpx.TransportError) as exc:
+                if not hops:
+                    raise FetchTransientError(str(exc)) from exc
+                hops.append(ChainHop(url, None))
+                return ExpandedChain(hops, url, None, False)
+            try:
+                status = resp.status_code
+                hops.append(ChainHop(str(parsed), status))
+                if status in _REDIRECT_STATUSES:
+                    location = resp.headers.get("location")
+                    if not location:
+                        return ExpandedChain(hops, str(parsed), status, False)
+                    url = str(parsed.join(location))
+                    continue
+                return ExpandedChain(hops, str(parsed), status, False)
+            finally:
+                await resp.aclose()
+    return ExpandedChain(hops, url, None, True)

middleware/rate_limiter.py

@@ -105,8 +105,14 @@ class Limits:
     # guessing is theater at 256 bits — this is belt-and-suspenders.
     URL_CLAIM = "30 per minute; 500 per day"
 
-    # Destination metadata fetch — outbound fetches on our dime; tight.
-    METADATA_FETCH = "20 per minute; 500 per day"
+    # Destination metadata fetch — outbound fetches on our dime, but the
+    # ~1h result cache means only novel URLs actually fetch.
+    METADATA_FETCH = "60 per minute; 2000 per day"
+
+    # Anonymous callers on GET /metadata (the link preview checker tool).
+    # Generous on purpose — it's a free tool — but still per-IP bounded:
+    # results cache ~1h, so only novel URLs cost an outbound fetch.
+    METADATA_ANON = "15 per minute; 300 per day"
 
     # Custom domains. Create counts FAILED attempts too (slowapi increments
     # at route entry), so the budget must absorb typos, blocked TLDs, and