#!/usr/bin/env python3
"""Offline ticket-attachment handoff with fictional observations and bytes.

No HTTP client, credentials, DNS, file writes, Zendesk mutations or ERP transfer.
URL syntax/origin checks are a bounded local policy, not an SSRF security audit.
"""
from dataclasses import dataclass, replace, asdict
import hashlib
import json
import re
from urllib.parse import urlsplit


@dataclass(frozen=True)
class Job:
    account: str
    ticket: int
    comment: int
    attachment: int
    erp_dossier: str


@dataclass(frozen=True)
class Metadata:
    job: Job
    content_url: str
    deleted: bool
    ticket_access_checked: bool
    collection_qualified: bool


@dataclass(frozen=True)
class Policy:
    account: str
    source_origin: str
    download_origins: tuple[str, ...]
    maximum_bytes: int


@dataclass(frozen=True)
class Reply:
    requested_url: str
    status: int
    location: str | None
    transport_qualified: bool


@dataclass(frozen=True)
class DownloadPlan:
    job: Job
    target_url: str
    authorization: None = None
    cookies: tuple = ()
    follow_redirects: bool = False


def positive(value):
    return type(value) is int and value > 0


def text(value):
    return isinstance(value, str) and bool(value.strip())


def origin(url):
    """Strict syntactic HTTPS profile; no DNS or connection is attempted."""
    if (not isinstance(url, str) or not url
            or any(ord(c) <= 32 or ord(c) == 127 for c in url)
            or "\\" in url or "#" in url):
        return None
    try:
        parsed = urlsplit(url)
        if (parsed.scheme != "https" or parsed.username is not None
                or parsed.password is not None or parsed.port not in (None, 443)
                or not parsed.hostname
                or re.fullmatch(r"[a-z0-9]+(?:[.-][a-z0-9]+)*", parsed.hostname) is None):
            return None
    except ValueError:
        return None
    return "https://" + parsed.hostname


def source_plan(job, metadata, policy):
    held = lambda reason: {"decision": "held", "reason": reason}
    if (not text(job.account) or not text(job.erp_dossier)
            or not all(positive(x) for x in (job.ticket, job.comment, job.attachment))):
        return held("job_outside_profile")
    if metadata.job != job or job.account != policy.account:
        return held("scope_mismatch")
    if metadata.collection_qualified is not True or metadata.ticket_access_checked is not True:
        return held("access_or_collection_unqualified")
    if metadata.deleted is not False:
        return held("attachment_not_confirmed_present")
    if origin(metadata.content_url) != policy.source_origin:
        return held("source_origin_outside_policy")
    if not urlsplit(metadata.content_url).path.startswith("/attachments/token/"):
        return held("source_path_outside_profile")
    return {"decision": "request_source", "credential_reference": job.account,
            "url": metadata.content_url, "follow_redirects": False}


def redirect_plan(job, metadata, policy, reply):
    source = source_plan(job, metadata, policy)
    if source["decision"] != "request_source":
        return source
    held = lambda reason: {"decision": "held", "reason": reason}
    if reply.transport_qualified is not True or reply.requested_url != source["url"]:
        return held("reply_unqualified_or_unrelated")
    if reply.status == 401:
        return held("source_authentication_review")
    if reply.status == 403:
        return held("source_permission_review")
    if type(reply.status) is not int or reply.status != 302:
        return held("response_outside_private_302_profile")
    if origin(reply.location) not in policy.download_origins:
        return held("redirect_origin_outside_policy")
    if reply.location == metadata.content_url:
        return held("redirect_loop")
    return DownloadPlan(job, reply.location)


def inspect_fixture(job, plan, status, payload, declared_size, policy):
    """Inspect supplied in-memory bytes; never marks an ERP transfer complete."""
    held = lambda reason: {"decision": "held", "reason": reason}
    if not isinstance(plan, DownloadPlan) or plan.job != job:
        return held("download_plan_or_scope_missing")
    if job.account != policy.account or origin(plan.target_url) not in policy.download_origins:
        return held("download_policy_changed")
    if (plan.authorization is not None or plan.cookies
            or plan.follow_redirects is not False):
        return held("unexpected_download_authority")
    if status == 403:
        return held("download_access_requires_new_observation")
    if type(status) is not int or status != 200:
        return held("download_not_observed_successful")
    if (not isinstance(payload, bytes) or not payload
            or not positive(declared_size) or len(payload) != declared_size
            or not positive(policy.maximum_bytes) or len(payload) > policy.maximum_bytes):
        return held("bytes_outside_local_contract")
    return {"decision": "fixture_bytes_observed", "identity": asdict(job),
            "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest(),
            "erp_transfer_confirmed": False}


def run_checks():
    names = []

    def check(name, condition):
        if name in names or not condition:
            raise AssertionError(name)
        names.append(name)

    job = Job("SUPPORT-A", 301, 401, 501, "ERP-DOSSIER-A")
    source = "https://support.example.invalid/attachments/token/fiction/?name=preuve.txt"
    target = "https://download.example.invalid/fixture?grant=fictitious"
    meta = Metadata(job, source, False, True, True)
    policy = Policy(job.account, "https://support.example.invalid",
                    ("https://download.example.invalid",), 1024)
    reply = Reply(source, 302, target, True)
    make = lambda j=job, m=meta, p=policy, r=reply: redirect_plan(j, m, p, r)
    plan = make()
    payload = b"Fictional support attachment, no customer data.\n"
    inspect = lambda j=job, p=plan, status=200, data=payload, size=len(payload), rules=policy: inspect_fixture(j, p, status, data, size, rules)
    receipt = inspect()
    check("authorized_source_is_separate_request", source_plan(job, meta, policy)["decision"] == "request_source")
    check("source_automatic_redirects_disabled", source_plan(job, meta, policy)["follow_redirects"] is False)
    check("approved_redirect_produces_plan", isinstance(plan, DownloadPlan))
    check("download_has_no_authorization", plan.authorization is None)
    check("download_has_no_source_cookies", plan.cookies == ())
    check("download_does_not_follow_another_redirect", plan.follow_redirects is False)
    check("other_account_isolated", make(j=replace(job, account="SUPPORT-B"))["reason"] == "scope_mismatch")
    check("other_ticket_isolated", make(j=replace(job, ticket=302))["reason"] == "scope_mismatch")
    check("other_comment_isolated", make(j=replace(job, comment=402))["reason"] == "scope_mismatch")
    check("other_attachment_isolated", make(j=replace(job, attachment=502))["reason"] == "scope_mismatch")
    check("other_erp_dossier_isolated", make(j=replace(job, erp_dossier="ERP-DOSSIER-B"))["reason"] == "scope_mismatch")
    check("boolean_ticket_is_not_identity", make(j=replace(job, ticket=True))["reason"] == "job_outside_profile")
    check("blank_dossier_has_no_destination", make(j=replace(job, erp_dossier=" "))["reason"] == "job_outside_profile")
    check("browser_session_not_access_proof", make(m=replace(meta, ticket_access_checked=False))["reason"] == "access_or_collection_unqualified")
    check("unqualified_metadata_held", make(m=replace(meta, collection_qualified=False))["reason"] == "access_or_collection_unqualified")
    check("deleted_attachment_held", make(m=replace(meta, deleted=True))["reason"] == "attachment_not_confirmed_present")
    check("unknown_presence_held", make(m=replace(meta, deleted=None))["reason"] == "attachment_not_confirmed_present")
    check("external_content_url_gets_no_source_plan", make(m=replace(meta, content_url=target))["reason"] == "source_origin_outside_policy")
    check("source_lookalike_domain_held", make(m=replace(meta, content_url=source.replace('support.example.invalid','support.example.invalid.evil.invalid')))["reason"] == "source_origin_outside_policy")
    check("other_attachment_surface_outside_profile", make(m=replace(meta, content_url="https://support.example.invalid/api/v2/attachments/501"))["reason"] == "source_path_outside_profile")
    check("unqualified_reply_held", make(r=replace(reply, transport_qualified=False))["reason"] == "reply_unqualified_or_unrelated")
    check("unrelated_source_reply_held", make(r=replace(reply, requested_url=target))["reason"] == "reply_unqualified_or_unrelated")
    check("source_401_is_authentication_review", make(r=replace(reply, status=401))["reason"] == "source_authentication_review")
    check("source_403_is_not_expiry_proof", make(r=replace(reply, status=403))["reason"] == "source_permission_review")
    for status in (200, 301, 404, 429, 500):
        check("source_status_"+str(status)+"_outside_profile", make(r=replace(reply, status=status))["reason"] == "response_outside_private_302_profile")
    for label, url in (
        ("missing", None), ("relative", "/download/file"),
        ("http", target.replace("https:", "http:")),
        ("userinfo", target.replace("https://", "https://user:pass@")),
        ("fragment", target+"#token"), ("control", target+"\n"),
        ("other_port", target.replace(".invalid/", ".invalid:8443/")),
        ("lookalike", target.replace("download.example.invalid", "download.example.invalid.evil.invalid")),
        ("unapproved", "https://other.example.invalid/file")):
        check("target_"+label+"_held", make(r=replace(reply, location=url))["reason"] == "redirect_origin_outside_policy")
    loop_policy = replace(policy, download_origins=policy.download_origins+(policy.source_origin,))
    check("source_redirect_loop_held", make(p=loop_policy, r=replace(reply, location=source))["reason"] == "redirect_loop")
    check("fixture_bytes_receive_local_digest", receipt["sha256"] == hashlib.sha256(payload).hexdigest())
    check("receipt_retains_attachment_identity", receipt["identity"]["attachment"] == 501)
    check("receipt_retains_erp_dossier", receipt["identity"]["erp_dossier"] == job.erp_dossier)
    check("receipt_is_not_erp_transfer_success", receipt["erp_transfer_confirmed"] is False)
    check("another_attachment_cannot_reuse_plan", inspect(j=replace(job, attachment=502))["reason"] == "download_plan_or_scope_missing")
    check("missing_download_plan_held", inspect(p=None)["reason"] == "download_plan_or_scope_missing")
    check("another_account_policy_held", inspect(rules=replace(policy, account="SUPPORT-B"))["reason"] == "download_policy_changed")
    check("withdrawn_download_origin_held", inspect(rules=replace(policy, download_origins=()))["reason"] == "download_policy_changed")
    check("download_403_requires_new_observation", inspect(status=403)["reason"] == "download_access_requires_new_observation")
    check("download_302_not_treated_as_file", inspect(status=302)["reason"] == "download_not_observed_successful")
    check("empty_bytes_held", inspect(data=b"", size=0)["reason"] == "bytes_outside_local_contract")
    check("truncated_fixture_held", inspect(data=payload[:-1])["reason"] == "bytes_outside_local_contract")
    check("size_mismatch_held", inspect(size=len(payload)+1)["reason"] == "bytes_outside_local_contract")
    check("local_size_limit_held", inspect(rules=replace(policy, maximum_bytes=8))["reason"] == "bytes_outside_local_contract")
    check("unexpected_source_authority_in_plan_held", inspect(p=replace(plan, authorization="do-not-forward"))["reason"] == "unexpected_download_authority")
    check("unexpected_cookie_in_plan_held", inspect(p=replace(plan, cookies=("session",)))["reason"] == "unexpected_download_authority")
    check("input_observations_unchanged", meta.content_url == source and reply.location == target)
    return {"checks_passed": len(names), "check_names": names,
            "scope": {"http_calls": 0, "zendesk_mutations": 0, "erp_operations": 0,
                      "storage_writes": 0, "dns_queries": 0,
                      "native_authorization_certified": False,
                      "download_lifetime_inferred": False},
            "examples": {"source403": make(r=replace(reply, status=403)),
                         "download403": inspect(status=403), "receipt": receipt,
                         "download_origin": origin(plan.target_url),
                         "download_authorization": plan.authorization}}


if __name__ == "__main__":
    print(json.dumps(run_checks(), ensure_ascii=False, indent=2))