src/gcmd_classifier/datasets/cmr.py

"""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"<form",
            b'type="password"',
            b"type='password'",
            b"sign in",
            b"log in",
            b"login",
        )
    )


def _raise_retrieval_error(exc: httpx.HTTPError) -> 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",
        )
    )