akif/arclet-boards

Arclet Boards — private Kanban workspaces with AppID, DataK3, shared web and MCP actions.

A remix template · 1 remixes · Remix akif/arclet-boards

tests/test_doors.py

import json
from uuid import uuid4

from fastapi.testclient import TestClient
import pytest

from app.main import create_app
from app.infrastructure.identity import person_from
from tests.support import OWNER, MEMBER, VIEWER, OUTSIDER, call, fixture_board, person_headers


@pytest.fixture
def web(monkeypatch):
    monkeypatch.setenv("DODIL_USER_POOL", "test/arclet-users")
    service, workspace, board = fixture_board()
    with TestClient(create_app(service)) as client:
        yield client, service, workspace, board


def invoke(client, tool, args=None, person=OWNER):
    headers = {**person_headers(person), "accept": "application/json, text/event-stream"}
    response = client.post("/mcp", headers=headers, json={
        "jsonrpc":"2.0", "id":2, "method":"tools/call",
        "params":{"name":tool,"arguments":args or {}}})
    assert response.status_code == 200, response.text
    return response.json()["result"]


def test_anonymous_api_and_mcp_are_401_but_discovery_and_health_are_public(web):
    client, _, _, _ = web
    assert client.get("/api/boards").status_code == 401
    assert client.post("/mcp", json={}).status_code == 401
    assert 'resource_metadata="https://testserver/.well-known/oauth-protected-resource/mcp"' in client.post("/mcp").headers["www-authenticate"]
    metadata = client.get("/.well-known/oauth-protected-resource/mcp").json()
    assert metadata["authorization_servers"] == ["https://appid.dodil.io/test/arclet-users"]
    assert metadata["resource"] == "https://testserver/mcp"
    assert client.get("/healthz").status_code == 200


def test_browser_redirects_to_gateway_and_does_not_ship_dev_login(web, monkeypatch):
    client, _, _, _ = web
    monkeypatch.setenv("DEV_USER", OWNER.email)
    response = client.get("/", follow_redirects=False)
    assert response.status_code == 303
    assert response.headers["location"] == "/.dodil/auth/login?return_to=/"
    assert client.get("/api/session").status_code == 401


def test_valid_gateway_person_can_read_session_and_assets(web):
    client, _, _, _ = web
    assert client.get("/api/session", headers=person_headers()).json()["sub"] == OWNER.sub
    response = client.get("/", headers=person_headers())
    assert response.status_code == 200 and 'Arclet Boards' in response.text
    assert "script-src 'self'" in response.headers["content-security-policy"]
    assert client.get("/static/app.js", headers=person_headers()).status_code == 200


def test_platform_identity_cannot_act_as_end_user(web):
    client, _, _, _ = web
    forged = {**person_headers(), "x-dodil-auth-source":"platform"}
    assert client.get("/api/boards", headers=forged).status_code == 401
    assert client.post("/mcp", headers=forged).status_code == 401


def test_user_header_without_verified_gateway_token_is_refused(web):
    client, _, _, _ = web
    assert client.get("/api/boards", headers={"x-dodil-user":json.dumps({"sub":OWNER.sub,"email":OWNER.email})}).status_code == 401


@pytest.mark.parametrize("claims", [
    {"amr":["client_credentials"]}, {"amr":[]}, {"iss":"https://attacker.test"},
    {"aud":"pool:other"}, {"exp":1}, {"sub":"different-sub"},
    {"email":"different@arclet.test"}, {"connection":"service"},
])
def test_gateway_claim_consistency_and_human_authentication_are_required(web, claims):
    client, _, _, _ = web
    assert client.get("/api/session", headers=person_headers(**claims)).status_code == 401


def test_web_mutation_calls_same_use_case_and_rejects_viewer(web):
    client, service, _, board = web
    args = {"board_id":board["id"],"list_id":board["lists"][0]["id"],"title":"From web"}
    response = client.post("/api/actions/add_card", headers={**person_headers(MEMBER),"Idempotency-Key":uuid4().hex}, json={"arguments":args})
    assert response.status_code == 200, response.text
    assert call(service, "get_board", board_id=board["id"])["cards"][0]["title"] == "From web"
    assert client.post("/api/actions/add_card", headers={**person_headers(VIEWER),"Idempotency-Key":uuid4().hex}, json={"arguments":args}).status_code == 403


def test_outsider_http_reads_and_writes_return_404(web):
    client, _, _, board = web
    assert client.get(f'/api/boards/{board["id"]}', headers=person_headers(OUTSIDER)).status_code == 404
    assert client.post("/api/actions/create_list", headers={**person_headers(OUTSIDER),"Idempotency-Key":uuid4().hex}, json={"arguments":{"board_id":board["id"],"title":"No"}}).status_code == 404


def test_csrf_cross_origin_write_is_refused(web):
    client, _, _, _ = web
    headers = {**person_headers(),"origin":"https://evil.test","Idempotency-Key":uuid4().hex}
    assert client.post("/api/actions/create_workspace", headers=headers, json={"arguments":{"title":"No"}}).status_code == 403


def test_gateway_public_host_is_used_for_oauth_and_same_origin_browser_writes(web):
    client, _, _, _ = web
    forwarding = {"host":"internal-service:8080","x-forwarded-host":"boards.ignite.dodil.cloud"}
    metadata = client.get("/.well-known/oauth-protected-resource/mcp",headers=forwarding).json()
    assert metadata["resource"] == "https://boards.ignite.dodil.cloud/mcp"
    response = client.post("/api/actions/create_workspace",headers={**forwarding,**person_headers(),
        "origin":"https://boards.ignite.dodil.cloud","cookie":"session=test","Idempotency-Key":uuid4().hex},
        json={"arguments":{"title":"From the real browser origin"}})
    assert response.status_code == 200


def test_http_rejects_missing_request_key_and_identity_spoofing_in_payload(web):
    client, _, _, _ = web
    assert client.post("/api/actions/create_workspace", headers=person_headers(), json={"arguments":{"title":"No"}}).status_code == 422
    assert client.post("/api/actions/create_workspace", headers={**person_headers(),"Idempotency-Key":uuid4().hex}, json={"arguments":{"title":"No","user_sub":OWNER.sub}}).status_code == 422


def test_mcp_tool_catalog_is_action_oriented_and_never_accepts_identity(web):
    client, _, _, _ = web
    response = client.post("/mcp", headers={**person_headers(),"accept":"application/json, text/event-stream"}, json={"jsonrpc":"2.0","id":1,"method":"tools/list"})
    assert response.status_code == 200, response.text
    tools = response.json()["result"]["tools"]
    names = {tool["name"] for tool in tools}
    assert {"list_boards","move_card","assign_card","set_due_date","invite_member"} <= names
    for tool in tools:
        assert not {"user_sub","actor","user","person","headers","ctx"} & set(tool["inputSchema"].get("properties",{}))
        assert not any(word in tool["name"] for word in ("sql","table","query"))
    move = next(tool for tool in tools if tool["name"] == "move_card")
    assert {"card_id","list_id","version","board_version","request_id"} <= set(move["inputSchema"]["required"])


def test_mcp_member_creates_real_application_card(web):
    client, service, _, board = web
    result = invoke(client,"add_card",{"board_id":board["id"],"list_id":board["lists"][0]["id"],"title":"From MCP","request_id":uuid4().hex},MEMBER)
    assert not result.get("isError"), result
    assert call(service,"get_board",board_id=board["id"])["cards"][0]["title"] == "From MCP"


def test_mcp_list_results_have_structured_json_arrays(web):
    client, _, _, board = web
    result = invoke(client,"list_boards",person=VIEWER)
    assert isinstance(result["structuredContent"]["result"],list)
    assert result["structuredContent"]["result"][0]["id"] == board["id"]


def test_mcp_viewer_read_succeeds_but_mutation_is_denied(web):
    client, _, _, board = web
    assert not invoke(client,"get_board",{"board_id":board["id"]},VIEWER).get("isError")
    result = invoke(client,"add_card",{"board_id":board["id"],"list_id":board["lists"][0]["id"],"title":"No","request_id":uuid4().hex},VIEWER)
    assert result.get("isError") and "403" in str(result)


def test_mcp_outsider_cannot_read_or_mutate(web):
    client, _, _, board = web
    assert invoke(client,"get_board",{"board_id":board["id"]},OUTSIDER).get("isError")
    assert invoke(client,"create_list",{"board_id":board["id"],"title":"No","request_id":uuid4().hex},OUTSIDER).get("isError")


def test_mcp_payload_cannot_override_authenticated_subject(web):
    client, service, _, board = web
    result = invoke(client,"add_card",{"board_id":board["id"],"list_id":board["lists"][0]["id"],"title":"No","request_id":uuid4().hex,"user_sub":OWNER.sub},VIEWER)
    assert result.get("isError")
    assert call(service,"get_board",board_id=board["id"])["cards"] == []