"""Strict public CMR collection retrieval and blind-boundary construction.""" from __future__ import annotations import copy import hashlib import json import logging import time from collections.abc import Callable from contextlib import closing from datetime import UTC, datetime from typing import Any from urllib.parse import parse_qsl, urljoin, urlsplit import httpx from pydantic import ValidationError from gcmd_classifier.config import DatasetCMRSettings from gcmd_classifier.datasets.errors import ( CMRError, CMRResponseError, CMRRetrievalError, CMRURLValidationError, ) from gcmd_classifier.datasets.models import ( BlindCollectionView, CMRPreclassificationFailure, CMRRedirectProvenance, CMRRetrievedSource, DatasetErrorRecord, DatasetIdentity, ResolvedCMRCollection, SealedExpertKeywordPayload, ) from gcmd_classifier.logging_config import get_logger, log_event _REDIRECT_STATUSES = {301, 302, 303, 307, 308} _APPROVED_RESPONSE_HEADERS = { "content-type", "content-length", "etag", "last-modified", "cmr-request-id", } _AUTH_QUERY_NAMES = { "access_token", "api_key", "apikey", "auth", "authorization", "bearer", "client_secret", "credential", "jwt", "password", "signature", "sig", "token", } class ValidatedCMRURL: """Validated URL and exact decoded Native ID.""" __slots__ = ("native_id", "url") def __init__(self, *, url: str, native_id: str) -> None: self.url = url self.native_id = native_id def validate_cmr_url( url: str, settings: DatasetCMRSettings | None = None, ) -> ValidatedCMRURL: """Validate a CMR collection-search URL without normalizing it.""" policy = settings or DatasetCMRSettings() if not isinstance(url, str) or not url: raise CMRURLValidationError("CMR_URL_INVALID", "CMR URL must be a non-empty string.") try: parsed = urlsplit(url) port = parsed.port except ValueError as exc: raise CMRURLValidationError("CMR_URL_INVALID", "CMR URL is malformed.") from exc if parsed.scheme != "https": raise CMRURLValidationError("CMR_URL_SCHEME", "CMR URL must use HTTPS.") if parsed.hostname != policy.host: raise CMRURLValidationError("CMR_URL_HOST", "CMR URL host is not approved.") if port not in (None, 443): raise CMRURLValidationError("CMR_URL_PORT", "CMR URL must use the default HTTPS port.") if parsed.username is not None or parsed.password is not None: raise CMRURLValidationError( "CMR_URL_USERINFO", "CMR URL must not contain user information." ) if parsed.path != policy.collection_path: raise CMRURLValidationError("CMR_URL_PATH", "CMR URL path is not approved.") if parsed.fragment: raise CMRURLValidationError("CMR_URL_FRAGMENT", "CMR URL must not contain a fragment.") try: query_items = parse_qsl(parsed.query, keep_blank_values=True, strict_parsing=False) except ValueError as exc: raise CMRURLValidationError("CMR_URL_QUERY", "CMR URL query is malformed.") from exc for name, _ in query_items: normalized_name = name.strip().lower().replace("-", "_") if normalized_name in _AUTH_QUERY_NAMES or any( marker in normalized_name for marker in ("token", "password", "credential", "authorization") ): raise CMRURLValidationError( "CMR_URL_AUTH_QUERY", "CMR URL contains a prohibited authentication-related query parameter.", ) native_values = [value for name, value in query_items if name == "native_id"] if len(native_values) != 1: raise CMRURLValidationError( "CMR_URL_NATIVE_ID_COUNT", "CMR URL must contain exactly one native_id query parameter.", ) if native_values[0] == "": raise CMRURLValidationError( "CMR_URL_NATIVE_ID_EMPTY", "CMR URL native_id query value must not be empty.", ) return ValidatedCMRURL(url=url, native_id=native_values[0]) def _safe_response_metadata(headers: httpx.Headers) -> dict[str, str]: return { name.lower(): value for name, value in headers.items() if name.lower() in _APPROVED_RESPONSE_HEADERS } def _is_json_media_type(value: str | None) -> bool: if not value: return False media_type = value.split(";", 1)[0].strip().lower() return media_type == "application/json" or ( media_type.startswith("application/") and media_type.endswith("+json") ) def _looks_like_login_response(content_type: str | None, body: bytes) -> bool: sample = body[:8192].lower() media_type = (content_type or "").lower() if "text/html" not in media_type and not sample.lstrip().startswith(b"<"): return False return any( marker in sample for marker in ( b" None: if isinstance(exc, httpx.TimeoutException): raise CMRRetrievalError("CMR_TIMEOUT", "CMR request timed out.") from exc if isinstance(exc, httpx.TransportError): raise CMRRetrievalError( "CMR_TRANSPORT_ERROR", "CMR connection, TLS, or transport failed.", ) from exc raise CMRRetrievalError("CMR_HTTP_ERROR", "CMR request failed.") from exc def retrieve_cmr_source( url: str, *, client: httpx.Client, settings: DatasetCMRSettings | None = None, utc_now: Callable[[], datetime] | None = None, monotonic: Callable[[], float] | None = None, ) -> tuple[ValidatedCMRURL, CMRRetrievedSource]: """Retrieve exact CMR source bytes using injected transport and manual redirects.""" policy = settings or DatasetCMRSettings() validated = validate_cmr_url(url, policy) now = utc_now or (lambda: datetime.now(UTC)) elapsed_clock = monotonic or time.monotonic started = elapsed_clock() current_url = validated.url visited = {current_url} redirects: list[CMRRedirectProvenance] = [] timeout = httpx.Timeout( connect=policy.connect_timeout_seconds, read=policy.read_timeout_seconds, write=policy.read_timeout_seconds, pool=policy.connect_timeout_seconds, ) headers = { "Accept": "application/json", "Accept-Encoding": "identity", "Client-Id": policy.client_id, } while True: if elapsed_clock() - started > policy.operation_timeout_seconds: raise CMRRetrievalError( "CMR_OPERATION_TIMEOUT", "CMR retrieval exceeded the overall operation limit.", ) try: request = client.build_request( "GET", current_url, headers=headers, timeout=timeout, ) if "authorization" in request.headers or "cookie" in request.headers: raise CMRRetrievalError( "CMR_CLIENT_STATE", "CMR requests must not include authorization or cookie state.", ) with closing( client.send( request, stream=True, follow_redirects=False, auth=None, ) ) as response: status = response.status_code if status in _REDIRECT_STATUSES: location = response.headers.get("location") if not location: raise CMRRetrievalError( "CMR_REDIRECT_LOCATION", "CMR redirect did not provide a Location header.", ) target = urljoin(current_url, location) target_validated = validate_cmr_url(target, policy) if target_validated.native_id != validated.native_id: raise CMRURLValidationError( "CMR_REDIRECT_NATIVE_ID", "CMR redirect changed the submitted Native ID.", ) if target in visited: raise CMRRetrievalError( "CMR_REDIRECT_LOOP", "CMR redirect loop was detected.", ) if len(redirects) >= policy.max_redirects: raise CMRRetrievalError( "CMR_REDIRECT_LIMIT", "CMR redirect limit was exceeded.", ) redirects.append( CMRRedirectProvenance( status_code=status, source_url=current_url, target_url=target, ) ) visited.add(target) current_url = target continue if 300 <= status < 400: raise CMRRetrievalError( "CMR_REDIRECT_UNSUPPORTED", "CMR returned an unsupported redirect status.", ) if status in {401, 403}: raise CMRRetrievalError( "CMR_NOT_PUBLIC", "CMR resource is not publicly accessible.", ) if status < 200 or status >= 300: raise CMRRetrievalError( "CMR_HTTP_STATUS", f"CMR returned HTTP status {status}.", ) content_length = response.headers.get("content-length") if content_length is not None: try: declared_length = int(content_length) except ValueError as exc: raise CMRRetrievalError( "CMR_CONTENT_LENGTH", "CMR response Content-Length is invalid.", ) from exc if declared_length > policy.max_response_bytes: raise CMRRetrievalError( "CMR_RESPONSE_TOO_LARGE", "CMR response exceeds the configured size limit.", ) content_encoding = response.headers.get("content-encoding") if content_encoding and content_encoding.lower() != "identity": raise CMRRetrievalError( "CMR_CONTENT_ENCODING", "CMR response used an unsupported content encoding.", ) chunks: list[bytes] = [] byte_count = 0 response_chunks = ( (response.content,) if response.is_stream_consumed else response.iter_raw() ) for chunk in response_chunks: if elapsed_clock() - started > policy.operation_timeout_seconds: raise CMRRetrievalError( "CMR_OPERATION_TIMEOUT", "CMR retrieval exceeded the overall operation limit.", ) byte_count += len(chunk) if byte_count > policy.max_response_bytes: raise CMRRetrievalError( "CMR_RESPONSE_TOO_LARGE", "CMR response exceeds the configured size limit.", ) chunks.append(chunk) body = b"".join(chunks) content_type = response.headers.get("content-type") if _looks_like_login_response(content_type, body): raise CMRRetrievalError( "CMR_LOGIN_RESPONSE", "CMR returned an authentication or login response.", ) if not _is_json_media_type(content_type): raise CMRRetrievalError( "CMR_MEDIA_TYPE", "CMR response does not use a supported JSON media type.", ) try: source_text = body.decode("utf-8") except UnicodeDecodeError as exc: raise CMRResponseError( "CMR_TEXT_DECODE", "CMR JSON response is not valid UTF-8.", ) from exc try: parsed = json.loads(source_text) except json.JSONDecodeError as exc: raise CMRResponseError( "CMR_JSON_INVALID", "CMR response contains malformed JSON.", ) from exc if not isinstance(parsed, dict): raise CMRResponseError( "CMR_JSON_ROOT", "CMR response root must be a JSON object.", ) source = CMRRetrievedSource( submitted_url=url, final_url=current_url, redirect_history=tuple(redirects), retrieved_at=now().astimezone(UTC).isoformat().replace("+00:00", "Z"), status_code=status, response_metadata=_safe_response_metadata(response.headers), response_bytes=body, source_text=source_text, sha256=hashlib.sha256(body).hexdigest(), parsed_json=copy.deepcopy(parsed), ) return validated, source except CMRError: raise except httpx.HTTPError as exc: _raise_retrieval_error(exc) def _required_non_empty_string(container: dict[str, Any], key: str, label: str) -> str: value = container.get(key) if not isinstance(value, str) or not value: raise CMRResponseError( "CMR_REQUIRED_FIELD", f"CMR collection requires a non-empty {label}.", ) return value def _canonical_payload_hash(value: Any) -> str: serialized = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(serialized).hexdigest() def resolve_cmr_collection( validated_url: ValidatedCMRURL, source: CMRRetrievedSource, ) -> ResolvedCMRCollection: """Resolve one collection and construct separate blind and sealed objects.""" root = source.parsed_json hits = root.get("hits") items = root.get("items") if isinstance(hits, bool) or not isinstance(hits, int) or hits < 0: raise CMRResponseError("CMR_HITS_INVALID", "CMR hits must be a non-negative integer.") if not isinstance(items, list): raise CMRResponseError("CMR_ITEMS_INVALID", "CMR items must be an array.") if hits != len(items): raise CMRResponseError( "CMR_CARDINALITY_MISMATCH", "CMR hits and item cardinality do not agree.", ) if len(items) != 1: raise CMRResponseError( "CMR_CARDINALITY", "CMR response must contain exactly one collection item.", ) item = items[0] if not isinstance(item, dict): raise CMRResponseError("CMR_ITEM_INVALID", "CMR collection item must be an object.") meta = item.get("meta") umm = item.get("umm") if not isinstance(meta, dict): raise CMRResponseError("CMR_META_INVALID", "CMR collection meta must be an object.") if not isinstance(umm, dict): raise CMRResponseError("CMR_UMM_INVALID", "CMR collection umm must be an object.") concept_id = _required_non_empty_string(meta, "concept-id", "Concept ID") returned_native_id = _required_non_empty_string(meta, "native-id", "Native ID") short_name = _required_non_empty_string(umm, "ShortName", "ShortName") version = _required_non_empty_string(umm, "Version", "Version") revision_id = meta.get("revision-id") if isinstance(revision_id, bool) or not isinstance(revision_id, int) or revision_id < 1: raise CMRResponseError( "CMR_REVISION_ID_INVALID", "CMR revision ID must be a positive integer.", ) derived_native_id = f"{short_name}_{version}" if validated_url.native_id != returned_native_id or returned_native_id != derived_native_id: raise CMRResponseError( "CMR_NATIVE_ID_MISMATCH", "Submitted, returned, and derived Native IDs do not match exactly.", ) related_urls = umm.get("RelatedUrls", []) if not isinstance(related_urls, list) or any( not isinstance(entry, dict) for entry in related_urls ): raise CMRResponseError( "CMR_RELATED_URLS_INVALID", "CMR RelatedUrls must be an array of objects when present.", ) expert_keywords = umm.get("ScienceKeywords", []) if not isinstance(expert_keywords, list) or any( not isinstance(entry, dict) for entry in expert_keywords ): raise CMRResponseError( "CMR_SCIENCE_KEYWORDS_INVALID", "CMR ScienceKeywords must be an array of objects when present.", ) try: identity = DatasetIdentity( concept_id=concept_id, native_id=returned_native_id, short_name=short_name, version=version, cmr_revision_id=revision_id, ) except ValidationError as exc: raise CMRResponseError( "CMR_IDENTITY_INVALID", "CMR collection identity is invalid.", ) from exc blind_material = { "identity": identity.model_dump(mode="json"), "derived_native_id": derived_native_id, "entry_title": umm.get("EntryTitle") if isinstance(umm.get("EntryTitle"), str) else None, "summary": umm.get("Abstract") if isinstance(umm.get("Abstract"), str) else None, "related_urls": copy.deepcopy(related_urls), "expert_keywords_excluded": True, } blind_view = BlindCollectionView( **blind_material, blind_view_sha256=_canonical_payload_hash(blind_material), ) sealed_keywords = copy.deepcopy(expert_keywords) sealed_material = { "dataset_key": concept_id, "source_path": "umm.ScienceKeywords", "science_keywords": sealed_keywords, } sealed_payload = SealedExpertKeywordPayload( dataset_key=concept_id, science_keywords=tuple(sealed_keywords), payload_sha256=_canonical_payload_hash(sealed_material), ) return ResolvedCMRCollection( dataset_key=concept_id, submitted_native_id=validated_url.native_id, returned_native_id=returned_native_id, derived_native_id=derived_native_id, identity=identity, source=source, source_item=copy.deepcopy(item), blind_view=blind_view, sealed_expert_keywords=sealed_payload, ) def retrieve_and_resolve_cmr_collection( url: str, *, client: httpx.Client, settings: DatasetCMRSettings | None = None, utc_now: Callable[[], datetime] | None = None, monotonic: Callable[[], float] | None = None, logger: logging.Logger | None = None, ) -> ResolvedCMRCollection: """Retrieve and resolve one collection while logging only safe metadata.""" active_logger = logger or get_logger("gcmd_classifier.datasets.cmr") try: validated, source = retrieve_cmr_source( url, client=client, settings=settings, utc_now=utc_now, monotonic=monotonic, ) result = resolve_cmr_collection(validated, source) except CMRError as exc: log_event( active_logger, "cmr_failed", stage="source", error_code=exc.code, ) raise log_event( active_logger, "cmr_resolved", stage="source", concept_id=result.dataset_key, native_id=result.identity.native_id, revision_id=result.identity.cmr_revision_id, source_sha256=result.source.sha256, status_code=result.source.status_code, ) return result def preclassification_failure(error: CMRError) -> CMRPreclassificationFailure: """Represent a CMR failure with explicit zero-inference semantics.""" return CMRPreclassificationFailure( error=DatasetErrorRecord( code=error.code, message=str(error), stage="source", ) )