# SPDX-License-Identifier: AGPL-3.0-only
# Copyright (c) 2026 sol pbc

from __future__ import annotations

import urllib.parse
from pathlib import Path
from typing import Any

import pytest
from flask import Flask
from werkzeug.routing import BuildError, Rule

from solstone.convey.secure_listener.wsgi import (
    CERTLESS_PAIR_ENDPOINTS,
    certless_target_allowed,
)
from solstone.think.link.nonces import NonceStore
from solstone.think.link.paths import nonces_path
from tests.link.certless_helpers import (
    certless_identity,
    dispatch_request,
    make_convey_app,
)


def test_route_enumeration_allows_only_pair_post(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app, _journal = make_convey_app(tmp_path, monkeypatch, link={"posture": "spl"})

    seen_allowed_pair_posts = set()
    for rule in app.url_map.iter_rules():
        for method in _route_methods(rule):
            path = _path_for_rule(app, rule, method)
            allowed = certless_target_allowed(app, path, method)
            expected = rule.endpoint in CERTLESS_PAIR_ENDPOINTS and method == "POST"
            assert allowed is expected, (rule.endpoint, method, path)
            if expected:
                seen_allowed_pair_posts.add(rule.endpoint)

    assert seen_allowed_pair_posts == set(CERTLESS_PAIR_ENDPOINTS)
    assert certless_target_allowed(app, "/app/network/pair", "POST") is True
    assert certless_target_allowed(app, "/app/network/pair", "GET") is False
    assert certless_target_allowed(app, "/app/link/pair", "POST") is True
    assert certless_target_allowed(app, "/app/link/pair", "GET") is False


def test_named_non_pair_endpoints_are_refused(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    app, _journal = make_convey_app(tmp_path, monkeypatch, link={"posture": "spl"})
    endpoints = {
        "app:network.pair_start": "POST",
        "app:network.api_status": "GET",
        "app:link.pair_start": "POST",
        "app:link.api_status": "GET",
        "app:observer.ingest_upload": "POST",
        "app:observer.ingest_event": "POST",
        "app:observer.ingest_segments": "GET",
        "app:observer.ingest_manifest": "GET",
        "app:observer.ingest_manifest_day": "GET",
        "app:import.journal_source_manifest": "GET",
        "app:import.ingest_segments": "POST",
        "app:import.ingest_entities": "POST",
        "app:import.ingest_facets": "POST",
        "app:import.ingest_imports": "POST",
        "app:import.ingest_config": "POST",
        "root.callosum_sse": "GET",
        "static": "GET",
    }

    for endpoint, method in endpoints.items():
        path = _path_for_endpoint(app, endpoint, method)
        assert certless_target_allowed(app, path, method) is False, (
            endpoint,
            method,
            path,
        )

    for path, method in (
        ("/", "GET"),
        ("/sse/events", "GET"),
        ("/app/network/api/devices", "GET"),
        ("/app/link/api/devices", "GET"),
        ("/api/config", "GET"),
    ):
        assert certless_target_allowed(app, path, method) is False


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["pl-via-spl", "pl-direct"])
async def test_evasion_paths_are_refused(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    mode: str,
) -> None:
    app, _journal = make_convey_app(tmp_path, monkeypatch, link={"posture": "spl"})
    NonceStore(nonces_path()).add("live", "phone")

    for prefix in ("/app/network", "/app/link"):
        for raw_path in (
            f"{prefix}/%2e%2e/pair",
            f"{prefix}/../pair-start",
            f"{prefix}/./pair",
            f"{prefix}/pair.",
            f"{prefix}/pair/.",
        ):
            path_info = urllib.parse.unquote(raw_path)
            assert certless_target_allowed(app, path_info, "POST") is False, raw_path

    for raw_path in ("/app/network%2Fpair", "/app/link%2Fpair"):
        response = await dispatch_request(
            app,
            certless_identity(mode),
            "POST",
            raw_path,
            body=b"{}",
            headers={"content-type": "application/json"},
        )

        assert response.status == 403
        assert b"pairing tunnel may only use /app/network/pair" in response.body


@pytest.mark.asyncio
@pytest.mark.parametrize("mode", ["pl-via-spl", "pl-direct"])
async def test_window_recheck_refuses_before_handler(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    mode: str,
) -> None:
    app, _journal = make_convey_app(tmp_path, monkeypatch, link={"posture": "spl"})
    called = False

    def pair_spy() -> str:
        nonlocal called
        called = True
        return "reached"

    for endpoint in CERTLESS_PAIR_ENDPOINTS:
        app.view_functions[endpoint] = pair_spy

    for path in ("/app/network/pair", "/app/link/pair"):
        response = await dispatch_request(
            app,
            certless_identity(mode),
            "POST",
            path,
            body=b"{}",
            headers={"content-type": "application/json"},
        )

        assert response.status == 403
        assert b"pairing window closed" in response.body
        assert called is False


def _route_methods(rule: Rule) -> list[str]:
    return sorted((rule.methods or set()) - {"HEAD", "OPTIONS"})


def _path_for_endpoint(app: Flask, endpoint: str, method: str) -> str:
    rules = [
        rule
        for rule in app.url_map.iter_rules(endpoint)
        if method in (rule.methods or set())
    ]
    assert rules, f"missing route {endpoint} {method}"
    return _path_for_rule(app, rules[0], method)


def _path_for_rule(app: Flask, rule: Rule, method: str) -> str:
    adapter = app.url_map.bind("solstone.local", url_scheme="https")
    try:
        built = adapter.build(
            rule.endpoint,
            _values_for_rule(rule),
            method=method,
            force_external=False,
        )
    except BuildError as exc:
        raise AssertionError(f"could not build {rule.endpoint} {method}") from exc
    return urllib.parse.urlsplit(built).path


def _values_for_rule(rule: Rule) -> dict[str, Any]:
    return {argument: _dummy_value(rule, argument) for argument in rule.arguments}


def _dummy_value(rule: Rule, argument: str) -> object:
    converter = rule._converters.get(argument)
    converter_name = converter.__class__.__name__ if converter is not None else ""
    if converter_name in {"IntegerConverter", "FloatConverter"}:
        return 1
    if argument in {"day", "date"}:
        return "20260413"
    if argument == "month":
        return "202604"
    if argument == "filename":
        return "app.css"
    if argument in {"key", "key_prefix"}:
        return "deadbeef"
    if converter_name == "PathConverter":
        return "path/value"
    return "value"
