21743a08b1
Sweep-the-edges hardening release (roadmap item #4). Audits every write-shaped backend endpoint to confirm each one enforces an explicit auth.require_contributor (or stricter) gate before doing state-changing work; adds a regression test net (test_anon_offlimits_vertical.py, 12 tests, 66 assertions) so future endpoints can't quietly ship without a gate. The only behaviour change: GET /api/rfcs/<slug>/graduate/progress now requires auth.require_user since the step detail (repo name, PR number, rollback steps) isn't part of the v0.3.0 anonymous-read contract. No schema, no env, no dependency changes; operator action is rebuild + restart per CHANGELOG §0.6.0 upgrade steps. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
477 lines
18 KiB
Python
477 lines
18 KiB
Python
"""v0.6.0 (roadmap item #4) — "anon discuss + contribute off-limits"
|
|
vertical.
|
|
|
|
A sweep-the-edges hardening release. The v0.3.0 release hid the write
|
|
affordances from anonymous viewers; v0.5.0 added the PR-less discussion
|
|
surface with its own write gate. v0.6.0 audits both: every write-shaped
|
|
endpoint refuses anonymous callers with 401 (or 403 when the role check
|
|
runs after the auth check), and every anonymous-read surface stays
|
|
reachable.
|
|
|
|
This test is the regression net for the audit. It walks each module's
|
|
representative write endpoint as an anonymous client and asserts the
|
|
401/403, then walks the same surfaces' representative read endpoints
|
|
as anonymous and asserts the 200. The intent is breadth over depth:
|
|
one assertion per write endpoint family is enough to catch a
|
|
regression where someone strips the `auth.require_contributor` line.
|
|
|
|
Endpoints covered (one or two from each module):
|
|
|
|
- api.py: propose, decline (admin), withdraw,
|
|
funder credentials POST/DELETE, funder consent
|
|
POST/DELETE
|
|
- api_branches.py: promote-to-branch, start-edit-branch, metadata,
|
|
manual-flush, visibility, grants POST/DELETE,
|
|
threads POST, thread messages POST, resolve,
|
|
chat-seen, change accept/decline/reask
|
|
- api_prs.py: pr-draft, open-pr, seen, review, merge, withdraw,
|
|
description, resolution-branch
|
|
- api_discussion.py: thread create, message post, resolve
|
|
- api_admin.py: role POST, mute POST, allowlist POST/DELETE
|
|
- api_notifications.py: prefs POST, watch POST, mark-read POST,
|
|
quiet-hours POST, user-mute POST/DELETE
|
|
- api_graduation.py: graduate POST, claim POST, progress GET
|
|
|
|
The §15.7 reads (`/api/notifications`, `/api/watches`,
|
|
`/api/users/me/*`) are per-user surfaces — they require an
|
|
authenticated viewer by definition; an anonymous 401 on those reads is
|
|
shape-correct, not a regression. The test does not assert reads on
|
|
those.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
# Reuse the fixture / session / fake-Gitea harness from Slice 1.
|
|
from test_propose_vertical import ( # noqa: F401 — fixtures land via import
|
|
FakeGitea,
|
|
app_with_fake_gitea,
|
|
provision_user_row,
|
|
sign_in_as,
|
|
tmp_env,
|
|
)
|
|
from test_rfc_view_vertical import SEED_BODY, seed_active_rfc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_anonymous_can_read_every_public_surface(app_with_fake_gitea):
|
|
"""Per §14 / the v0.3.0 anonymous-read contract: the catalog, the
|
|
RFC view, the PR-less discussion surface, the philosophy page, and
|
|
the health probe must remain reachable for unauthenticated viewers.
|
|
This is the read side of the item #4 contract — the read surfaces
|
|
must NOT regress to require auth as the write gates tighten.
|
|
"""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
provision_user_row(user_id=1, login="alice", role="contributor")
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
|
|
# No session cookie — viewer is anonymous.
|
|
client.cookies.clear()
|
|
|
|
# The five read surfaces an anonymous viewer must reach.
|
|
assert client.get("/api/health").status_code == 200
|
|
assert client.get("/api/philosophy").status_code == 200
|
|
assert client.get("/api/auth/me").status_code == 200
|
|
assert client.get("/api/rfcs").status_code == 200
|
|
assert client.get("/api/rfcs/ohm").status_code == 200
|
|
assert client.get("/api/rfcs/ohm/main").status_code == 200
|
|
assert client.get("/api/rfcs/ohm/discussion/threads").status_code == 200
|
|
assert client.get("/api/proposals").status_code == 200
|
|
|
|
|
|
def test_anonymous_propose_refused(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
client.cookies.clear()
|
|
r = client.post(
|
|
"/api/rfcs/propose",
|
|
json={"title": "X", "slug": "x", "pitch": "p", "tags": []},
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_anonymous_proposal_admin_paths_refused(app_with_fake_gitea):
|
|
"""The admin-gated proposal actions — merge, decline — must refuse
|
|
anonymous callers with 401 (the auth check runs before the role
|
|
check; both refusals are correct, but 401 is the structural signal
|
|
"no session at all")."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
client.cookies.clear()
|
|
# PR number doesn't need to exist — the gate runs first.
|
|
assert client.post("/api/proposals/1/merge").status_code == 401
|
|
assert (
|
|
client.post("/api/proposals/1/decline", json={"comment": "no"}).status_code
|
|
== 401
|
|
)
|
|
assert client.post("/api/proposals/1/withdraw").status_code == 401
|
|
|
|
|
|
def test_anonymous_branch_writes_refused_on_active_rfc(app_with_fake_gitea):
|
|
"""Branch-scoped writes on an active RFC: promote-to-branch,
|
|
manual-flush, visibility, grants, threads create, message post,
|
|
resolve, chat-seen, change accept/decline/reask. All must 401 for
|
|
anonymous callers."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
client.cookies.clear()
|
|
|
|
# Branch-scoped writes — slug + branch values are placeholders;
|
|
# the auth gate runs before any state lookup.
|
|
slug = "ohm"
|
|
branch = "feature-x"
|
|
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/main/promote-to-branch",
|
|
json={},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/manual-flush",
|
|
json={"new_content": "hi", "paragraph_count": 1},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/visibility",
|
|
json={"read_public": False},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/grants",
|
|
json={"grantee_gitea_login": "alice"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.delete(
|
|
f"/api/rfcs/{slug}/branches/{branch}/grants/alice",
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/threads",
|
|
json={"thread_kind": "chat", "anchor_kind": "whole-doc"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/threads/1/messages",
|
|
json={"text": "hi"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/threads/1/resolve",
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/chat-seen",
|
|
json={"last_seen_message_id": 1},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/changes/1/accept",
|
|
json={"proposed": "x"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/changes/1/decline",
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/changes/1/reask",
|
|
).status_code == 401
|
|
)
|
|
# Chat stream — POST shaped, same auth gate.
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/threads/1/chat",
|
|
json={"text": "hi"},
|
|
).status_code == 401
|
|
)
|
|
|
|
|
|
def test_anonymous_super_draft_writes_refused(app_with_fake_gitea):
|
|
"""Super-draft-scoped writes: start-edit-branch and metadata. The
|
|
PR open / merge paths share the gate via api_prs.py — see the
|
|
PR-flow test below for those."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
client.cookies.clear()
|
|
assert (
|
|
client.post(
|
|
"/api/rfcs/anything/start-edit-branch", json={}
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/rfcs/anything/metadata", json={"title": "x"}
|
|
).status_code == 401
|
|
)
|
|
|
|
|
|
def test_anonymous_pr_flow_writes_refused(app_with_fake_gitea):
|
|
"""All §10 PR-flow writes — open, merge, withdraw, description,
|
|
review, seen, pr-draft, resolution-branch — must 401 for anonymous."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
client.cookies.clear()
|
|
slug, branch, pr = "ohm", "feature-x", 1
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/pr-draft"
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/branches/{branch}/open-pr",
|
|
json={"title": "t", "description": "d"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/prs/{pr}/seen",
|
|
json={"last_seen_message_id": 1},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/prs/{pr}/review",
|
|
json={"text": "x", "anchor_payload": {}},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/prs/{pr}/merge"
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/prs/{pr}/withdraw"
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/prs/{pr}/description",
|
|
json={"title": "t", "description": "d"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
f"/api/rfcs/{slug}/prs/{pr}/resolution-branch"
|
|
).status_code == 401
|
|
)
|
|
|
|
|
|
def test_anonymous_discussion_writes_refused(app_with_fake_gitea):
|
|
"""The v0.5.0 PR-less discussion surface — write gates must hold.
|
|
This duplicates the assertion in `test_discussion_vertical.py` and
|
|
keeps it here too as the canonical home for the item #4 audit."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
client.cookies.clear()
|
|
assert (
|
|
client.post(
|
|
"/api/rfcs/ohm/discussion/threads",
|
|
json={"message": "drive-by"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/rfcs/ohm/discussion/threads/1/messages",
|
|
json={"text": "drive-by"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/rfcs/ohm/discussion/threads/1/resolve"
|
|
).status_code == 401
|
|
)
|
|
|
|
|
|
def test_anonymous_admin_writes_refused(app_with_fake_gitea):
|
|
"""Admin surfaces — role, mute, allowlist — refuse anonymous.
|
|
The auth check runs before the require_admin role check, so the
|
|
response is 401."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
client.cookies.clear()
|
|
assert (
|
|
client.post(
|
|
"/api/admin/users/1/role", json={"role": "admin"}
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/admin/users/1/mute", json={"muted": True}
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/admin/allowlist", json={"email": "x@y.z"}
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.delete("/api/admin/allowlist/x@y.z").status_code == 401
|
|
)
|
|
# Admin reads also gated.
|
|
assert client.get("/api/admin/users").status_code == 401
|
|
assert client.get("/api/admin/audit").status_code == 401
|
|
assert client.get("/api/admin/permission-events").status_code == 401
|
|
assert client.get("/api/admin/graduation-queue").status_code == 401
|
|
assert client.get("/api/admin/allowlist").status_code == 401
|
|
|
|
|
|
def test_anonymous_notification_writes_refused(app_with_fake_gitea):
|
|
"""Notification preference / watch / mark-read / user-mute writes —
|
|
all per-user surfaces, all require an authenticated viewer."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
client.cookies.clear()
|
|
assert (
|
|
client.post(
|
|
"/api/users/me/notification-preferences",
|
|
json={"email_personal_direct": False},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/api/users/me/quiet-hours",
|
|
json={"start": None, "end": None, "timezone": None},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post("/api/rfcs/ohm/watch", json={"state": "watching"}).status_code
|
|
== 401
|
|
)
|
|
assert client.post("/api/notifications/1/read").status_code == 401
|
|
assert (
|
|
client.post("/api/notifications/read", json={}).status_code == 401
|
|
)
|
|
assert client.post("/api/users/1/notification-mute").status_code == 401
|
|
assert client.delete("/api/users/1/notification-mute").status_code == 401
|
|
|
|
|
|
def test_anonymous_funder_writes_refused(app_with_fake_gitea):
|
|
"""§6.7 funder credential + consent writes — registering a key,
|
|
consenting to fund — all refuse anonymous callers."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
client.cookies.clear()
|
|
assert (
|
|
client.post(
|
|
"/api/users/me/funder/credentials",
|
|
json={"provider": "anthropic", "api_key": "sk-test"},
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.delete(
|
|
"/api/users/me/funder/credentials/anthropic"
|
|
).status_code == 401
|
|
)
|
|
assert (
|
|
client.post("/api/rfcs/ohm/funder/consent").status_code == 401
|
|
)
|
|
assert (
|
|
client.delete("/api/rfcs/ohm/funder/consent").status_code == 401
|
|
)
|
|
|
|
|
|
def test_anonymous_graduation_writes_refused(app_with_fake_gitea):
|
|
"""§13 graduation: the POST kickoff and POST claim both refuse
|
|
anonymous. The progress SSE was gated to require_user in v0.6.0
|
|
(item #4) since it surfaces admin-internal step detail."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
client.cookies.clear()
|
|
assert (
|
|
client.post(
|
|
"/api/rfcs/anything/graduate",
|
|
json={
|
|
"rfc_id": "RFC-0001",
|
|
"repo_name": "rfc-0001-x",
|
|
"owners": ["alice"],
|
|
},
|
|
).status_code == 401
|
|
)
|
|
assert client.post("/api/rfcs/anything/claim").status_code == 401
|
|
# v0.6.0 tightening: progress SSE now requires require_user.
|
|
# No graduation is in flight, but the auth check runs first.
|
|
assert (
|
|
client.get("/api/rfcs/anything/graduate/progress").status_code == 401
|
|
)
|
|
|
|
|
|
def test_anonymous_can_read_published_pr_view(app_with_fake_gitea):
|
|
"""The PR review page is §11.3 universal-public — once a PR is
|
|
open, anonymous viewers can read it. This guards against a
|
|
regression where the read endpoint accidentally grows an auth
|
|
gate."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
|
|
# Seed an open PR row directly — the cache shape is enough for
|
|
# the read endpoint; the live Gitea fetch falls back gracefully.
|
|
db.conn().execute(
|
|
"""
|
|
INSERT INTO cached_prs
|
|
(rfc_slug, pr_kind, repo, pr_number, title, description, state,
|
|
opened_by, opened_at, head_branch, base_branch, head_sha)
|
|
VALUES ('ohm', 'rfc_branch', 'wiggleverse/rfc-0001-ohm', 7, 't', 'd',
|
|
'open', 'alice', datetime('now'), 'feature-x', 'main', 'sha7')
|
|
"""
|
|
)
|
|
client.cookies.clear()
|
|
# Anonymous read on an open PR: should be 200. The endpoint may
|
|
# surface a partial response (the FakeGitea won't have the head
|
|
# branch's RFC.md, so branch_body falls back to empty) but the
|
|
# auth gate must let the read through.
|
|
r = client.get("/api/rfcs/ohm/prs/7")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["capabilities"]["is_anonymous"] is True
|
|
assert body["capabilities"]["can_merge"] is False
|
|
assert body["capabilities"]["can_post_review"] is False
|