jarca-ledger-mcp/scripts/persistent_auth_http_evidence.py

#!/usr/bin/env python3
"""Exercise persistent-auth HTTP responses without printing secrets or tokens."""

from __future__ import annotations

import argparse
import http.client
import json
import os
import ssl
from urllib.parse import urlencode, urlsplit

from streamlit.proto.BackMsg_pb2 import BackMsg
from streamlit.proto.ClientState_pb2 import ClientState
from streamlit.proto.ForwardMsg_pb2 import ForwardMsg
from websockets.sync.client import connect

COOKIE_NAME = "__Host-jarca_dashboard_auth"


def _args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--base-url", default="https://127.0.0.1:18506")
    return parser.parse_args()


def _connection(parsed):
    if parsed.scheme != "https" or parsed.hostname not in {"127.0.0.1", "localhost"}:
        raise SystemExit("Evidence target must be loopback HTTPS")
    context = ssl.create_default_context()
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE
    return http.client.HTTPSConnection(parsed.hostname, parsed.port, context=context, timeout=10)


def _request(parsed, method: str, path: str, *, body: bytes | None = None, headers=None):
    connection = _connection(parsed)
    connection.request(method, path, body=body, headers=headers or {})
    response = connection.getresponse()
    payload = response.read()
    result = response.status, {key.lower(): value for key, value in response.getheaders()}, payload
    connection.close()
    return result


def _assert_streamlit_restore(parsed, cookie_pair: str) -> int:
    context = ssl.create_default_context()
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE
    origin = f"https://{parsed.hostname}:{parsed.port}"
    websocket_url = f"wss://{parsed.hostname}:{parsed.port}/_stcore/stream"
    with connect(
        websocket_url,
        ssl=context,
        origin=origin,
        subprotocols=["streamlit"],
        additional_headers={"Cookie": cookie_pair},
        proxy=None,
        open_timeout=30,
    ) as websocket:
        back_message = BackMsg()
        back_message.rerun_script.CopyFrom(
            ClientState(query_string="page=%EC%9A%94%EC%95%BD")
        )
        websocket.send(back_message.SerializeToString())
        messages: list[str] = []
        for _ in range(100):
            raw = websocket.recv(timeout=15)
            if isinstance(raw, str):
                continue
            forward_message = ForwardMsg()
            forward_message.ParseFromString(raw)
            messages.append(str(forward_message))
            if forward_message.HasField("script_finished"):
                break
    combined = "\n".join(messages)
    if "이번 달 수입/지출 현황" not in combined or "관리자 로그인" in combined:
        raise SystemExit("Signed cookie did not restore the authenticated Streamlit page")
    return len(messages)


def main() -> None:
    parsed = urlsplit(_args().base_url)
    admin_id = os.environ["JARCA_DASHBOARD_ADMIN_ID"]
    password = os.environ["JARCA_DASHBOARD_ADMIN_PASSWORD"]
    origin = f"https://{parsed.hostname}:{parsed.port}"

    body = urlencode(
        {"admin_id": admin_id, "admin_password": password, "remember_me": "on"}
    ).encode()
    status, headers, _ = _request(
        parsed,
        "POST",
        "/api/auth/login",
        body=body,
        headers={
            "Content-Type": "application/x-www-form-urlencoded",
            "Content-Length": str(len(body)),
            "Origin": origin,
        },
    )
    cookie_header = headers.get("set-cookie", "")
    required = [COOKIE_NAME + "=", "HttpOnly", "Secure", "SameSite=lax", "Path=/", "Max-Age=2592000"]
    if status != 303 or headers.get("location") != "/" or not all(
        marker in cookie_header for marker in required
    ):
        raise SystemExit("Login response did not satisfy the hardened cookie contract")
    cookie_pair = cookie_header.split(";", 1)[0]

    status, _, _ = _request(
        parsed,
        "GET",
        "/",
        headers={"Cookie": cookie_pair},
    )
    if status != 200:
        raise SystemExit("Authenticated reconnect request failed")
    websocket_messages = _assert_streamlit_restore(parsed, cookie_pair)

    status, headers, _ = _request(
        parsed,
        "POST",
        "/api/auth/logout",
        headers={"Origin": origin, "Cookie": cookie_pair, "Content-Length": "0"},
    )
    deletion = headers.get("set-cookie", "")
    if status != 303 or not all(
        marker in deletion
        for marker in [COOKIE_NAME + "=", "HttpOnly", "Secure", "SameSite=lax", "Path=/", "Max-Age=0"]
    ):
        raise SystemExit("Logout response did not satisfy the cookie deletion contract")

    print(
        json.dumps(
            {
                "success": True,
                "transport": "loopback HTTPS",
                "login_status": 303,
                "reconnect_status": 200,
                "streamlit_websocket_authenticated": True,
                "streamlit_messages": websocket_messages,
                "logout_status": 303,
                "cookie_name": COOKIE_NAME,
                "cookie_value": "[REDACTED]",
                "remember_me_max_age": 2592000,
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()