"""Public URL + Open Graph verification (stages 12 + 13). The `vps_deployment` stage writes the HTML; the `url_verification` and `og_verification` stages fetch the public URL back over HTTPS to assert: url_verification: - status == 200 - Content-Type starts with text/html - no `WWW-Authenticate` / no auth challenge header present og_verification: - all of: og:title / og:description / og:image / og:url are present in - og:type, og:site_name, twitter:card present - the og:image URL itself returns HTTP 200 + image/* content-type (publicly accessible) Synchronous (not async) by design — these run inside the orchestrator's per-show coroutine; using httpx sync keeps the failure surface small. The upstream calls are short (<10s timeout) and the failure semantics are early. """ from __future__ import annotations import logging from typing import Iterable import httpx from bs4 import BeautifulSoup logger = logging.getLogger(__name__) _USER_AGENT = "maxamini-brief-verifier/12 (+https://just2done.com)" _TIMEOUT = httpx.Timeout(15.0, connect=10.0) class VerificationError(RuntimeError): pass def verify_public_url( url: str, *, require_og: bool = False, og_image_url: str | None = None, ) -> None: """Raise `VerificationError` with a redacted reason if any assertion fails. No exception ⇒ assertions passed; the orchestrator marks the stage ok. """ try: with httpx.Client(timeout=_TIMEOUT, follow_redirects=True, headers={"User-Agent": _USER_AGENT}) as client: response = client.get(url) except httpx.HTTPError as exc: raise VerificationError(f"url fetch failed: {exc}") from exc if response.status_code != 200: raise VerificationError(f"url returned {response.status_code}, expected 200") content_type = response.headers.get("content-type", "") if "text/html" not in content_type.lower(): raise VerificationError(f"content-type is not text/html (got {content_type!r})") # No auth challenge should ever be returned for a public brief. if response.headers.get("www-authenticate"): raise VerificationError("response carried a WWW-Authenticate header (auth required)") if not require_og: return # OG assertions follow. soup = BeautifulSoup(response.text, "html.parser") og_tags = {tag.get("property") or tag.get("name"): tag.get("content") for tag in soup.find_all("meta")} required = ("og:title", "og:description", "og:image", "og:url", "og:type", "og:site_name", "twitter:card") missing = [k for k in required if not og_tags.get(k)] if missing: raise VerificationError(f"missing OG tags: {', '.join(missing)}") og_image = og_image_url or og_tags.get("og:image") if not og_image: raise VerificationError("og:image missing content") try: with httpx.Client(timeout=_TIMEOUT, follow_redirects=True, headers={"User-Agent": _USER_AGENT}) as client: image_response = client.get(og_image) except httpx.HTTPError as exc: raise VerificationError(f"og:image fetch failed: {exc}") from exc if image_response.status_code != 200: raise VerificationError(f"og:image returned {image_response.status_code}") image_ct = image_response.headers.get("content-type", "") if not image_ct.lower().startswith("image/"): raise VerificationError(f"og:image content-type is not image/* (got {image_ct!r})") if image_response.headers.get("www-authenticate"): raise VerificationError("og:image response carried a WWW-Authenticate header")