astrololo/services/logic/tests/test_link_crypto.py

"""Szyfrowanie łącza między warstwami (PRE-16).

Sedno: przez to łącze płyną surowe wiersze oryginalnych baz interpretacyjnych.
Testy nie sprawdzają więc tylko, czy „coś się zaszyfrowało i odszyfrowało" —
sprawdzają, czy tajnego tekstu FAKTYCZNIE NIE MA w bajtach lecących po sieci
oraz czy każda znana droga na skróty (brak szyfrowania, obcy klucz, podmieniony
bajt, przeklejenie na inny endpoint, odtworzenie po czasie) kończy się odmową.
"""
import json
import pathlib
import time

import pytest
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from starlette.testclient import TestClient

from app import link_crypto
from app.link_crypto import ENV_LOGIC_DATA, ENV_PRESENTATION_LOGIC, Link, LinkError

SECRET = "Saturn w VII domu — opis z oryginalnej bazy interpretacyjnej"
KEY_A = "11" * 32          # hex, 32 bajty
KEY_B = "22" * 32


class Payload(BaseModel):
    question: str


def _app(link: Link | None) -> FastAPI:
    app = FastAPI()
    if link is not None:
        app.add_middleware(link_crypto.LinkCryptoMiddleware, link=link, layer="testowa")

    @app.get("/health")
    def health():
        return {"status": "ok"}

    @app.post("/search")
    def search(payload: Payload):
        return {"echo": payload.question, "interpretation": SECRET}

    @app.get("/catalog")
    def catalog():
        return {"models": ["a", "b"]}

    @app.post("/stream")
    def stream():
        def lines():
            for i in range(4):
                yield json.dumps({"step": i, "note": SECRET}).encode() + b"\n"

        return StreamingResponse(lines(), media_type="application/x-ndjson")

    return app


@pytest.fixture
def link():
    return Link(link_crypto.parse_key(KEY_A))


# ------------------------------------------------------------------- klucze

def test_key_accepts_hex_and_base64():
    import base64

    raw = bytes(range(32))
    assert link_crypto.parse_key(raw.hex()) == raw
    assert link_crypto.parse_key(base64.b64encode(raw).decode()) == raw


def test_key_of_wrong_length_is_rejected_loudly():
    """Krótki klucz to nie „słabsze szyfrowanie", tylko błąd konfiguracji."""
    with pytest.raises(LinkError, match="32"):
        link_crypto.parse_key("aabb")


def test_key_that_is_neither_hex_nor_base64_is_rejected():
    with pytest.raises(LinkError, match="hex"):
        link_crypto.parse_key("!!! to nie jest klucz !!!")


def test_key_from_env_is_lazy(monkeypatch):
    monkeypatch.delenv(ENV_PRESENTATION_LOGIC, raising=False)
    assert link_crypto.key_from_env(ENV_PRESENTATION_LOGIC) is None
    monkeypatch.setenv(ENV_PRESENTATION_LOGIC, KEY_A)
    assert link_crypto.key_from_env(ENV_PRESENTATION_LOGIC) == bytes.fromhex(KEY_A)


def test_directions_use_different_subkeys(link):
    """Żądanie i odpowiedź nie dzielą klucza — powtórzenie jednorazówki w jedną
    stronę nie osłabia drugiej."""
    stamp = link_crypto.stamp_now()
    sealed = link.seal(link_crypto.REQUEST, "/search", stamp, 0, b"tajne")
    with pytest.raises(LinkError):
        link.open(link_crypto.RESPONSE, "/search", stamp, 0, sealed)


# --------------------------------------------------------- podstawowy obieg

def test_round_trip_delivers_plaintext_to_the_app(link):
    with TestClient(_app(link)) as client:
        got = link_crypto.call_json(client, "POST", "http://testserver/search",
                                    payload={"question": "Saturn"}, link=link)
    assert got["echo"] == "Saturn"
    assert got["interpretation"] == SECRET


def test_get_without_body_also_works(link):
    """GET nie ma ciała, ale i tak pieczętujemy pustą kopertę — to ona dowodzi,
    że pytający ma klucz, i ona wymusza zaszyfrowanie odpowiedzi."""
    with TestClient(_app(link)) as client:
        got = link_crypto.call_json(client, "GET", "http://testserver/catalog", link=link)
    assert got == {"models": ["a", "b"]}


def test_without_key_traffic_stays_plaintext(link):
    """Dev bez sekretów ma działać jak dotąd — inaczej nikt nie odpali projektu lokalnie."""
    with TestClient(_app(None)) as client:
        got = link_crypto.call_json(client, "POST", "http://testserver/search",
                                    payload={"question": "Saturn"}, link=None)
    assert got["interpretation"] == SECRET


def test_health_stays_open_for_kubernetes_probes(link):
    """Sondy k8s klucza nie mają. Gdyby /health wymagał szyfrowania, literówka
    w sekrecie kładłaby pody zamiast pokazać błąd w aplikacji."""
    with TestClient(_app(link)) as client:
        assert client.get("http://testserver/health").json() == {"status": "ok"}


# ------------------------------------------- czy na kablu naprawdę nic nie widać

def _raw_exchange(client, link, payload):
    """Wysyła zapieczętowane żądanie i zwraca SUROWE bajty obu stron."""
    stamp = link_crypto.stamp_now()
    body = link_crypto.frame_out(
        link.seal(link_crypto.REQUEST, "/search", stamp, 0, json.dumps(payload).encode()))
    response = client.request(
        "POST", "http://testserver/search", content=body,
        headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
                 link_crypto.HEADER_TS: stamp,
                 "Content-Type": link_crypto.CONTENT_TYPE})
    return body, response


def test_request_bytes_do_not_contain_the_question(link):
    with TestClient(_app(link)) as client:
        body, _ = _raw_exchange(client, link, {"question": "Saturn w VII"})
    assert b"Saturn" not in body
    assert b"question" not in body


def test_response_bytes_do_not_contain_the_interpretation(link):
    """To jest właściwy powód istnienia całego modułu."""
    with TestClient(_app(link)) as client:
        _, response = _raw_exchange(client, link, {"question": "Saturn"})
    assert response.status_code == 200
    assert SECRET.encode() not in response.content
    assert b"interpretation" not in response.content
    assert response.headers[link_crypto.HEADER_ENC] == link_crypto.VERSION
    assert response.headers["content-type"] == link_crypto.CONTENT_TYPE


# ------------------------------------------------------- drogi na skróty i ataki

def test_plaintext_request_is_refused_when_key_is_set(link):
    """Fail-closed: regresja po stronie klienta nie może oznaczać cichego
    powrotu do jawnego ruchu."""
    with TestClient(_app(link)) as client:
        response = client.post("http://testserver/search", json={"question": "Saturn"})
    assert response.status_code == 400
    assert SECRET.encode() not in response.content


def test_foreign_key_cannot_read_the_link(link):
    """Klucze są osobne dla każdej pary warstw — przejęcie jednego nie otwiera drugiej."""
    intruder = Link(link_crypto.parse_key(KEY_B))
    with TestClient(_app(link)) as client:
        stamp = link_crypto.stamp_now()
        body = link_crypto.frame_out(
            intruder.seal(link_crypto.REQUEST, "/search", stamp, 0, b'{"question":"x"}'))
        response = client.request(
            "POST", "http://testserver/search", content=body,
            headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
                     link_crypto.HEADER_TS: stamp,
                     "Content-Type": link_crypto.CONTENT_TYPE})
    assert response.status_code == 400
    assert SECRET.encode() not in response.content


def test_single_flipped_bit_is_rejected(link):
    """GCM uwierzytelnia, więc nie ma wariantu „odszyfrowało się, ale zmienione"."""
    with TestClient(_app(link)) as client:
        stamp = link_crypto.stamp_now()
        sealed = bytearray(link.seal(link_crypto.REQUEST, "/search", stamp, 0,
                                     b'{"question":"x"}'))
        sealed[-1] ^= 0x01
        response = client.request(
            "POST", "http://testserver/search",
            content=link_crypto.frame_out(bytes(sealed)),
            headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
                     link_crypto.HEADER_TS: stamp,
                     "Content-Type": link_crypto.CONTENT_TYPE})
    assert response.status_code == 400


def test_frame_cannot_be_replayed_against_another_endpoint(link):
    """Ścieżka wchodzi do materiału uwierzytelnianego, więc podsłuchanej koperty
    nie da się przekleić tam, gdzie odpowiedź byłaby ciekawsza."""
    stamp = link_crypto.stamp_now()
    sealed = link.seal(link_crypto.REQUEST, "/catalog", stamp, 0, b"")
    with pytest.raises(LinkError):
        link.open(link_crypto.REQUEST, "/search", stamp, 0, sealed)


def test_frames_cannot_be_reordered(link):
    """Numer ramki jest uwierzytelniony — przestawienie kolejności w strumieniu
    to błąd, a nie po cichu pomieszany horoskop."""
    stamp = link_crypto.stamp_now()
    second = link.seal(link_crypto.RESPONSE, "/stream", stamp, 1, b"druga")
    with pytest.raises(LinkError):
        link.open(link_crypto.RESPONSE, "/stream", stamp, 0, second)


def test_stale_frame_is_refused(link, monkeypatch):
    """Bez okna czasowego podsłuchane żądanie dałoby się odtworzyć kiedykolwiek."""
    old = f"{time.time() - link_crypto.MAX_SKEW_SECONDS - 60:.3f}"
    with TestClient(_app(link)) as client:
        body = link_crypto.frame_out(
            link.seal(link_crypto.REQUEST, "/search", old, 0, b'{"question":"x"}'))
        response = client.request(
            "POST", "http://testserver/search", content=body,
            headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
                     link_crypto.HEADER_TS: old,
                     "Content-Type": link_crypto.CONTENT_TYPE})
    assert response.status_code == 400


def test_truncated_stream_is_an_error_not_silent_loss(link):
    stamp = link_crypto.stamp_now()
    full = link_crypto.frame_out(link.seal(link_crypto.RESPONSE, "/x", stamp, 0, b"abc"))
    with pytest.raises(LinkError, match="urwana"):
        link.open_all(link_crypto.RESPONSE, "/x", stamp, full[:-2])


def test_absurd_frame_length_does_not_allocate(link):
    """Zadeklarowana długość pochodzi z sieci — nie wolno jej wierzyć na słowo."""
    import struct

    with pytest.raises(LinkError, match="rozmiar"):
        list(link_crypto.frames_in(struct.pack(">I", 2 ** 31) + b"nic"))


# ---------------------------------------------------------- odpowiedź strumieniowa

def test_streaming_response_survives_encryption(link):
    """Okno postępu dostaje kolejne linie na żywo — muszą dojść po kolei
    i w komplecie, mimo że każda jedzie w osobnej kopercie."""
    with TestClient(_app(link)) as client:
        stamp = link_crypto.stamp_now()
        body = link_crypto.frame_out(link.seal(link_crypto.REQUEST, "/stream", stamp, 0, b""))
        with client.stream("POST", "http://testserver/stream", content=body,
                           headers={link_crypto.HEADER_ENC: link_crypto.VERSION,
                                    link_crypto.HEADER_TS: stamp,
                                    "Content-Type": link_crypto.CONTENT_TYPE}) as response:
            chunks = list(link_crypto.open_response_stream(response, link))

    steps = [json.loads(line) for line in b"".join(chunks).splitlines()]
    assert [s["step"] for s in steps] == [0, 1, 2, 3]
    assert all(s["note"] == SECRET for s in steps)


def test_incremental_unframing_handles_split_frames(link):
    """Ramka potrafi rozjechać się między dwa odczyty z gniazda — składamy ją
    w buforze, zamiast zakładać, że każdy kawałek to komplet."""
    stamp = link_crypto.stamp_now()
    stream = b"".join(link.seal_stream(link_crypto.RESPONSE, "/x", stamp,
                                       [b"raz", b"dwa", b"trzy"]))
    buffer = bytearray()
    opened, seq = [], 0
    for i in range(0, len(stream), 5):            # ciachamy w poprzek ramek
        buffer += stream[i:i + 5]
        for frame in link_crypto.unframe_incremental(buffer):
            opened.append(link.open(link_crypto.RESPONSE, "/x", stamp, seq, frame))
            seq += 1
    assert opened == [b"raz", b"dwa", b"trzy"]
    assert not buffer, "bufor musi zostać pusty — inaczej gdzieś zgubiliśmy ramkę"


# ----------------------------------------------- trzy kopie muszą być identyczne

def test_all_three_services_share_the_same_module():
    """Moduł jest skopiowany do trzech niezależnych usług (nie mają wspólnej
    biblioteki). Rozjazd między kopiami objawiłby się dopiero na produkcji jako
    „nie da się odszyfrować" — więc pilnujemy tego testem."""
    root = pathlib.Path(__file__).resolve().parents[3]
    copies = {svc: (root / "services" / svc / "app" / "link_crypto.py")
              for svc in ("presentation", "logic", "data")}
    missing = [svc for svc, path in copies.items() if not path.is_file()]
    assert not missing, f"brak modułu w warstwach: {missing}"
    contents = {svc: path.read_bytes() for svc, path in copies.items()}
    assert len(set(contents.values())) == 1, (
        "kopie link_crypto.py rozjechały się między warstwami: "
        + ", ".join(f"{svc}={len(body)}B" for svc, body in contents.items())
    )


def test_env_names_are_two_distinct_keys():
    """Wymóg wprost: osobny klucz dla pary prezentacja-logika i logika-dane."""
    assert ENV_PRESENTATION_LOGIC != ENV_LOGIC_DATA


# ------------------------------------------------- klient też musi być fail-closed
#
# To wyszło dopiero z podsłuchu prawdziwego gniazda, nie z testów: przy kliencie
# BEZ klucza serwer owszem odmawiał, ale pytanie leciało po drodze otwartym
# tekstem. Odpowiedź była chroniona — zapytanie już nie.

def test_client_without_key_sends_nothing_when_encryption_required(monkeypatch):
    monkeypatch.setenv(link_crypto.ENV_REQUIRED, "true")
    sent = []

    class Tripwire:
        def request(self, *args, **kwargs):
            sent.append(args)
            raise AssertionError("żądanie NIE powinno opuścić procesu")

    with pytest.raises(LinkError, match=link_crypto.ENV_REQUIRED):
        link_crypto.call(Tripwire(), "POST", "http://logic/search",
                         payload={"value": "[Sat"}, link=None)
    assert not sent, "treść zapytania wyszłaby jawnym tekstem"


def test_plaintext_still_allowed_in_dev(monkeypatch, link):
    """Bez tej flagi lokalne uruchomienie bez sekretów ma dalej działać."""
    monkeypatch.delenv(link_crypto.ENV_REQUIRED, raising=False)
    with TestClient(_app(None)) as client:
        got = link_crypto.call_json(client, "POST", "http://testserver/search",
                                    payload={"question": "Saturn"}, link=None)
    assert got["interpretation"] == SECRET


def test_service_refuses_to_start_without_key_when_required(monkeypatch):
    """Pod w CrashLoop widać od razu; usługę, która wstała i nie szyfruje — nie."""
    monkeypatch.setenv(link_crypto.ENV_REQUIRED, "true")
    monkeypatch.delenv(ENV_LOGIC_DATA, raising=False)
    with pytest.raises(LinkError, match=ENV_LOGIC_DATA):
        link_crypto.install(FastAPI(), ENV_LOGIC_DATA, "danych")