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

scripts/verify_live.py

"""Verify the actual Ignite gateway with four real AppID user sessions.

Supply ARCLET_LIVE_URL and ARCLET_TOKEN_OWNER/MEMBER/VIEWER/OUTSIDER in the
environment. No forged identity headers stand in for authentication. Creates
private verification records owned by the supplied owner and records only
statuses/resource ids as evidence. Does not print or save tokens.
"""
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
import json
import os
from pathlib import Path
from uuid import uuid4

import httpx


def verify(base_url, tokens, evidence_path, platform_token=None, issuer=None):
    base_url = base_url.rstrip("/")
    checks = []
    client = httpx.Client(base_url=base_url, timeout=90, follow_redirects=False)

    def check(name, condition):
        if not condition:
            raise AssertionError(name)
        checks.append(name)
        print("PASS",name,flush=True)

    def headers(role):
        return {"Authorization":"Bearer " + tokens[role]}

    def request(role, method, path, **kwargs):
        return client.request(method,path,headers={**headers(role),**kwargs.pop("headers",{})},**kwargs)

    def action(role, name, arguments, expected=200, key=None):
        response = request(role,"POST","/api/actions/"+name,
            headers={"Idempotency-Key":key or uuid4().hex},json={"arguments":arguments})
        if response.status_code != expected:
            raise AssertionError(f"{name}: expected HTTP {expected}, got {response.status_code}: {response.text[:500]}")
        return response.json()

    def tool(role, name, arguments=None, error=False):
        response = request(role,"POST","/mcp",headers={"accept":"application/json, text/event-stream"},
            json={"jsonrpc":"2.0","id":1,"method":"tools/call",
                  "params":{"name":name,"arguments":arguments or {}}})
        if response.status_code != 200:
            raise AssertionError(f"MCP {name}: HTTP {response.status_code}: {response.text[:300]}")
        result = response.json()["result"]
        if bool(result.get("isError")) != error:
            raise AssertionError(f"MCP {name}: unexpected error result {result}")
        if error:
            return result
        if "structuredContent" in result:
            data = result["structuredContent"]
            return data.get("result",data)
        values = [json.loads(block["text"]) for block in result["content"] if block["type"] == "text"]
        if name in {"list_workspaces","list_boards","list_my_cards","list_overdue","get_card_activity","list_members"}:
            return values[0] if len(values) == 1 and isinstance(values[0],list) else values
        return values[0]

    sessions = {}
    for role in tokens:
        response = request(role,"GET","/api/session")
        check(f"real AppID {role} identity accepted by deployed gateway",response.status_code == 200)
        sessions[role] = response.json()
    check("four independent user subjects",len({row["sub"] for row in sessions.values()}) == 4)
    if platform_token:
        platform = {"Authorization":"Bearer "+platform_token}
        check("real platform principal cannot read end-user API",client.get("/api/boards",headers=platform).status_code == 401)
        check("real platform principal cannot invoke application MCP",client.post("/mcp",headers=platform,json={}).status_code == 401)
    check("anonymous board API denied",client.get("/api/boards").status_code == 401)
    mcp = client.post("/mcp",json={})
    check("anonymous MCP challenge advertises OAuth discovery",mcp.status_code == 401 and "resource_metadata=" in mcp.headers.get("www-authenticate",""))
    metadata = client.get("/.well-known/oauth-protected-resource/mcp")
    check("OAuth metadata is public and points to this MCP resource",metadata.status_code == 200 and metadata.json()["resource"] == base_url+"/mcp")
    expected_issuer = issuer or os.environ.get("ARCLET_EXPECTED_ISSUER","https://appid.dodil.io/akif/arclet-boards-users")
    check("OAuth authorization server is the application's AppID pool",metadata.json()["authorization_servers"] == [expected_issuer])
    forged = {"x-dodil-auth-source":"pool","x-dodil-user":json.dumps(sessions["owner"]),
              "x-dodil-user-jwt":"forged.token.value"}
    check("gateway strips forged anonymous identity headers",client.get("/api/boards",headers=forged).status_code == 401)
    actual = request("viewer","GET","/api/session",headers=forged)
    check("forged owner header cannot replace a real viewer session",actual.status_code == 200 and actual.json()["sub"] == sessions["viewer"]["sub"])

    workspace = action("owner","create_workspace",{"title":"Arclet verification workspace"})
    board = tool("owner","create_board",{"workspace_id":workspace["id"],"title":"Launch, with a little clarity",
        "description":"A real board used to verify Arclet's deployed collaboration flows.",
        "color":"indigo","request_id":uuid4().hex})
    check("MCP creates a persistent real board",bool(board.get("id")))
    bid = board["id"]
    for role in ("member","viewer"):
        tool("owner","invite_member",{"board_id":bid,"email":sessions[role]["email"],
            "role":role,"request_id":uuid4().hex})
    check("owner grants board-specific member and viewer access",True)
    get_board = lambda role="owner": request(role,"GET",f"/api/boards/{bid}").json()
    board = get_board()
    check("member and viewer read their accessible board",get_board("member")["role"] == "member" and get_board("viewer")["role"] == "viewer")
    check("outsider cannot discover private board",request("outsider","GET",f"/api/boards/{bid}").status_code == 404)
    check("outsider board listing excludes private board",bid not in {row["id"] for row in request("outsider","GET","/api/boards").json()})
    action("member","invite_member",{"board_id":bid,"email":sessions["outsider"]["email"]},403)
    action("viewer","create_list",{"board_id":bid,"title":"Forbidden"},403)
    check("only owner manages members and viewer cannot edit",True)
    card = tool("member","add_card",{"board_id":bid,"list_id":board["lists"][0]["id"],
        "title":"Prepare the launch notes","description":"Capture the purpose, audience, and next small step.",
        "request_id":uuid4().hex})
    check("member MCP mutation persists through HTTP",request("owner","GET",f'/api/cards/{card["id"]}').json()["title"] == card["title"])
    tool("viewer","add_card",{"board_id":bid,"list_id":board["lists"][0]["id"],"title":"Forbidden","request_id":uuid4().hex},error=True)
    tool("outsider","get_board",{"board_id":bid},error=True)
    tool("outsider","create_list",{"board_id":bid,"title":"Forbidden","request_id":uuid4().hex},error=True)
    check("MCP viewer mutation and outsider reads/writes are denied",True)
    tool("viewer","add_card",{"board_id":bid,"list_id":board["lists"][0]["id"],"title":"Spoofed", "user_sub":sessions["owner"]["sub"],"request_id":uuid4().hex},error=True)
    check("MCP user-identity arguments cannot elevate permissions",True)
    updated = action("member","update_card",{"card_id":card["id"],"title":card["title"],
        "description":"A thoughtful release starts with a clear story.","assignee":sessions["member"]["email"],
        "due_date":"2020-01-01","labels":[{"name":"Writing","color":"indigo"},{"name":"Priority","color":"rose"}],"version":card["version"]})
    check("description labels assignment and date persist",request("owner","GET",f'/api/cards/{card["id"]}').json()["labels"] == updated["labels"])
    check("my-tasks and overdue are scoped to the real member",card["id"] in {row["id"] for row in tool("member","list_my_cards")} and card["id"] in {row["id"] for row in tool("member","list_overdue")})
    board = get_board()
    moved = tool("member","move_card",{"card_id":card["id"],"list_id":board["lists"][1]["id"],
        "index":0,"version":updated["version"],"board_version":board["version"],"request_id":uuid4().hex})
    check("MCP cross-list move persists",request("owner","GET",f'/api/cards/{card["id"]}').json()["list_id"] == moved["list_id"])
    activity = tool("viewer","get_card_activity",{"card_id":card["id"]})
    check("history contains creation field changes labels and move with real actors",
          {"created","updated","labels_updated","moved"} <= {row["action"] for row in activity} and all(row["actor_sub"] in {sessions["owner"]["sub"],sessions["member"]["sub"]} for row in activity))
    args = {"card_id":card["id"],"due_date":"2026-12-15","version":moved["version"]}
    key = uuid4().hex
    first = action("member","set_due_date",args,key=key)
    check("durable HTTP replay returns original response",action("member","set_due_date",args,key=key) == first)
    action("member","assign_card",{"card_id":card["id"],"assignee":sessions["outsider"]["email"],"version":first["version"]},422)
    action("member","set_due_date",{"card_id":card["id"],"due_date":"2026-02-31","version":first["version"]},422)
    action("member","set_due_date",{"card_id":card["id"],"due_date":None,"version":moved["version"]},409)
    check("invalid assignee invalid date and stale edit are refused",True)
    item = action("member","add_checklist_item",{"card_id":card["id"],"title":"Review the story with the team","version":first["version"]})
    action("viewer","complete_checklist_item",{"item_id":item["id"],"completed":True,"version":item["version"]},403)
    action("member","complete_checklist_item",{"item_id":item["id"],"completed":True,"version":item["version"]})
    check("manual checklist persists and viewer cannot toggle it",request("owner","GET",f'/api/cards/{card["id"]}').json()["checklist"][0]["completed"] is True)
    other_workspace = action("outsider","create_workspace",{"title":"Isolated verification workspace"})
    other_board = action("outsider","create_board",{"workspace_id":other_workspace["id"],"title":"Private outsider board"})
    other = request("outsider","GET",f'/api/boards/{other_board["id"]}').json()
    board = get_board()
    fresh_card = request("member","GET",f'/api/cards/{card["id"]}').json()
    action("member","move_card",{"card_id":card["id"],"list_id":other["lists"][0]["id"],"index":0,
        "version":fresh_card["version"],"board_version":board["version"]},404)
    check("cross-board foreign list cannot receive a card",True)
    action("owner","change_member_role",{"board_id":bid,"email":sessions["member"]["email"],"role":"viewer"})
    action("member","create_list",{"board_id":bid,"title":"Forbidden after downgrade"},403)
    action("owner","change_member_role",{"board_id":bid,"email":sessions["member"]["email"],"role":"member"})
    check("role changes take effect without replacing the user's token",True)
    # A second private board allows a real removal check without spoiling the
    # main verification board used by the browser tests.
    removable = action("owner","create_board",{"workspace_id":workspace["id"],"title":"Membership removal verification"})
    action("owner","invite_member",{"board_id":removable["id"],"email":sessions["member"]["email"]})
    action("owner","remove_member",{"board_id":removable["id"],"email":sessions["member"]["email"]})
    check("revoked user loses visibility immediately",request("member","GET",f'/api/boards/{removable["id"]}').status_code == 404)
    with httpx.Client(base_url=base_url,timeout=90) as fresh_client:
        persisted = fresh_client.get(f'/api/cards/{card["id"]}',headers=headers("owner"))
        check("record survives new HTTP session and backend connection",persisted.status_code == 200 and persisted.json()["title"] == card["title"])
    report = {"checked_at":datetime.now(timezone.utc).isoformat(),"base_url":base_url,
        "authentication":"four real AppID password-grant sessions through the Ignite gateway",
        "passed":checks,"workspace_id":workspace["id"],"board_id":bid,"card_id":card["id"],
        "oauth_discovery_verified":True,"external_assistant_interactive_consent_verified":False}
    Path(evidence_path).write_text(json.dumps(report,indent=2)+"\n",encoding="utf-8")
    client.close()
    return report


if __name__ == "__main__":
    verify(os.environ["ARCLET_LIVE_URL"],
        {role:os.environ["ARCLET_TOKEN_"+role.upper()] for role in ("owner","member","viewer","outsider")},
        os.environ.get("ARCLET_EVIDENCE_PATH","evidence/live.json"))