Jonathan Kirkwood
3a4b7dfd8b
Pre-release audit fixes: - External Administrators can only act on investors already in their own funds (grant_access, create_statement, set_partner_exited, import match), closing a cross-fund account-takeover chain. - Role and is_active changes are gated: no self-escalation, non-approvers cannot mint or alter approver accounts, and the Service Admin is protected. - must_change_password is enforced server-side, not just in the client. - Session cookie is Secure with a 12h max-age and SameSite; CSP and standard security headers on every response; the app refuses to boot on a known or short session secret; uvicorn runs with proxy-headers so login throttling keys on the real client IP behind the platform proxy. - Import preview no longer confirms cross-fund investors; no-store on document downloads. - Adds backend/tests/test_security_hardening.py (15 tests); full suite 76 passing. - SECURITY-AUDIT.md records the swarm audit, findings, and remediation status.
257 lines
11 KiB
Python
257 lines
11 KiB
Python
"""Regression tests for the pre-release security audit fixes.
|
|
|
|
Covers three classes of finding:
|
|
- Cross-tenant investor takeover by an external Administrator (scope-check on every write that
|
|
can mint an access grant or a capital statement for an arbitrary investor).
|
|
- Internal privilege escalation via role/is_active mass-assignment on the user endpoints.
|
|
- Server-side enforcement of the forced password change (not just the browser modal).
|
|
"""
|
|
|
|
from datetime import date
|
|
|
|
from sqlmodel import select
|
|
|
|
from ten31portal.models import (
|
|
CapitalAccountStatement, Entity, EntityAccess, EntityType, User, UserRole,
|
|
)
|
|
from tests.conftest import make_user
|
|
|
|
|
|
def _login(client, username, password="password123"):
|
|
client.post("/api/auth/logout")
|
|
resp = client.post("/api/auth/login", json={"login": username, "password": password})
|
|
assert resp.status_code == 200, resp.text
|
|
return resp
|
|
|
|
|
|
def _two_funds(session):
|
|
"""Fund A (managed by `fundadmin`) and Fund B. lp_a is in A, lp_b is in B (out of scope)."""
|
|
fund_a = Entity(name="Fund A", type=EntityType.fund)
|
|
fund_b = Entity(name="Fund B", type=EntityType.fund)
|
|
session.add(fund_a)
|
|
session.add(fund_b)
|
|
session.commit()
|
|
|
|
admin = make_user(session, username="fundadmin", role=UserRole.fund_administrator,
|
|
name="Outside Administrator")
|
|
session.add(EntityAccess(user_id=admin.id, entity_id=fund_a.id))
|
|
lp_a = make_user(session, username="lp-a", role=UserRole.investor, name="LP Alpha")
|
|
session.add(EntityAccess(user_id=lp_a.id, entity_id=fund_a.id))
|
|
lp_b = make_user(session, username="lp-b", role=UserRole.investor, name="LP Beta")
|
|
session.add(EntityAccess(user_id=lp_b.id, entity_id=fund_b.id))
|
|
session.commit()
|
|
return fund_a, fund_b, admin, lp_a, lp_b
|
|
|
|
|
|
# --- P0-1: cross-tenant investor takeover ------------------------------------------------
|
|
|
|
def test_administrator_cannot_grant_out_of_scope_investor(client, session, approver):
|
|
"""The takeover chain: grant an out-of-fund investor access to my fund, which would pull
|
|
them into scope and unlock reset-password. The grant itself must be refused."""
|
|
fund_a, fund_b, admin, lp_a, lp_b = _two_funds(session)
|
|
_login(client, "fundadmin")
|
|
|
|
resp = client.put(f"/api/users/{lp_b.id}/access/{fund_a.id}")
|
|
assert resp.status_code == 403, resp.text
|
|
# No EntityAccess row was minted, so lp_b never entered the Administrator's scope.
|
|
rows = session.exec(
|
|
select(EntityAccess).where(
|
|
EntityAccess.user_id == lp_b.id, EntityAccess.entity_id == fund_a.id
|
|
)
|
|
).all()
|
|
assert rows == []
|
|
# And the follow-on reset-password stays blocked.
|
|
assert client.post(f"/api/users/{lp_b.id}/reset-password",
|
|
json={"password": "attacker-set-pw"}).status_code == 403
|
|
|
|
|
|
def test_administrator_cannot_create_statement_for_out_of_scope_investor(client, session, approver):
|
|
fund_a, fund_b, admin, lp_a, lp_b = _two_funds(session)
|
|
_login(client, "fundadmin")
|
|
|
|
resp = client.post("/api/capital-accounts", json={
|
|
"entity_id": fund_a.id,
|
|
"investor_user_id": lp_b.id,
|
|
"as_of_date": "2026-03-31",
|
|
"commitment_dollars": 1000, "beginning_balance_dollars": 0,
|
|
"contributions_dollars": 1000, "distributions_dollars": 0,
|
|
"ending_balance_dollars": 1000,
|
|
})
|
|
assert resp.status_code == 403, resp.text
|
|
assert session.exec(
|
|
select(CapitalAccountStatement).where(
|
|
CapitalAccountStatement.investor_user_id == lp_b.id
|
|
)
|
|
).all() == []
|
|
|
|
|
|
def test_administrator_cannot_mint_exit_grant_for_out_of_scope_investor(client, session, approver):
|
|
fund_a, fund_b, admin, lp_a, lp_b = _two_funds(session)
|
|
_login(client, "fundadmin")
|
|
|
|
resp = client.put(f"/api/entities/{fund_a.id}/partners/{lp_b.id}/exited",
|
|
json={"exited_on": "2026-06-30"})
|
|
assert resp.status_code == 403, resp.text
|
|
assert session.exec(
|
|
select(EntityAccess).where(
|
|
EntityAccess.user_id == lp_b.id, EntityAccess.entity_id == fund_a.id
|
|
)
|
|
).all() == []
|
|
|
|
|
|
def test_administrator_can_still_manage_in_scope_investor(client, session, approver):
|
|
"""Positive control: the legitimate in-scope path is untouched."""
|
|
fund_a, fund_b, admin, lp_a, lp_b = _two_funds(session)
|
|
_login(client, "fundadmin")
|
|
|
|
# lp_a is already in fund_a, so a statement and a password reset both succeed.
|
|
assert client.post("/api/capital-accounts", json={
|
|
"entity_id": fund_a.id, "investor_user_id": lp_a.id, "as_of_date": "2026-03-31",
|
|
"commitment_dollars": 500, "beginning_balance_dollars": 0,
|
|
"contributions_dollars": 500, "distributions_dollars": 0,
|
|
"ending_balance_dollars": 500,
|
|
}).status_code == 201
|
|
assert client.post(f"/api/users/{lp_a.id}/reset-password",
|
|
json={"password": "a-fresh-password"}).status_code == 200
|
|
|
|
|
|
# --- P0-2: internal privilege escalation via mass-assignment -----------------------------
|
|
|
|
def test_operations_cannot_self_promote_to_approver(client, session):
|
|
ops = make_user(session, username="ops", role=UserRole.operations)
|
|
_login(client, "ops")
|
|
|
|
resp = client.patch(f"/api/users/{ops.id}", json={"role": "approver"})
|
|
assert resp.status_code == 403, resp.text
|
|
session.refresh(ops)
|
|
assert ops.role == UserRole.operations
|
|
|
|
|
|
def test_operations_cannot_create_approver(client, session):
|
|
make_user(session, username="ops", role=UserRole.operations)
|
|
_login(client, "ops")
|
|
|
|
resp = client.post("/api/users", json={
|
|
"name": "Sneaky", "username": "sneaky", "password": "password123", "role": "approver",
|
|
})
|
|
assert resp.status_code == 403, resp.text
|
|
|
|
|
|
def test_operations_cannot_deactivate_service_admin(client, session):
|
|
make_user(session, username="ops", role=UserRole.operations)
|
|
svc = make_user(session, username="svc", role=UserRole.approver, is_service_admin=True)
|
|
_login(client, "ops")
|
|
|
|
resp = client.patch(f"/api/users/{svc.id}", json={"is_active": False})
|
|
assert resp.status_code == 403, resp.text
|
|
session.refresh(svc)
|
|
assert svc.is_active is True
|
|
|
|
|
|
def test_operations_cannot_disable_an_approver(client, session):
|
|
make_user(session, username="ops", role=UserRole.operations)
|
|
boss = make_user(session, username="boss", role=UserRole.approver)
|
|
_login(client, "ops")
|
|
|
|
resp = client.patch(f"/api/users/{boss.id}", json={"is_active": False})
|
|
assert resp.status_code == 403, resp.text
|
|
|
|
|
|
def test_approver_can_still_manage_roles(client, session):
|
|
"""Positive control: an approver retains full role management."""
|
|
approver = make_user(session, username="chief", role=UserRole.approver)
|
|
ops = make_user(session, username="ops", role=UserRole.operations)
|
|
_login(client, "chief")
|
|
|
|
assert client.patch(f"/api/users/{ops.id}", json={"role": "approver"}).status_code == 200
|
|
session.refresh(ops)
|
|
assert ops.role == UserRole.approver
|
|
|
|
|
|
def test_operations_can_still_edit_ordinary_fields(client, session):
|
|
"""Non-privileged edits by an internal admin keep working."""
|
|
make_user(session, username="ops", role=UserRole.operations)
|
|
viewer = make_user(session, username="v", role=UserRole.viewer, name="Old Name")
|
|
_login(client, "ops")
|
|
|
|
assert client.patch(f"/api/users/{viewer.id}", json={"name": "New Name"}).status_code == 200
|
|
session.refresh(viewer)
|
|
assert viewer.name == "New Name"
|
|
|
|
|
|
# --- P0-3: server-side forced password change --------------------------------------------
|
|
|
|
def test_must_change_password_blocks_api_until_rotated(client, session):
|
|
make_user(session, username="lp", role=UserRole.investor, password="Ten31Portal",
|
|
must_change_password=True)
|
|
resp = client.post("/api/auth/login", json={"login": "lp", "password": "Ten31Portal"})
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
# /me is allowed (the client needs it to learn it must prompt), but data endpoints are not.
|
|
assert client.get("/api/auth/me").status_code == 200
|
|
assert client.get("/api/entities").status_code == 403
|
|
assert client.get("/api/documents").status_code == 403
|
|
|
|
# Rotating the password lifts the block.
|
|
changed = client.post("/api/auth/change-password", json={
|
|
"current_password": "Ten31Portal", "new_password": "my-own-strong-pw",
|
|
})
|
|
assert changed.status_code == 200, changed.text
|
|
assert client.get("/api/entities").status_code == 200
|
|
|
|
|
|
# --- P1/P2: transport and header hardening -----------------------------------------------
|
|
|
|
def test_security_headers_present(client):
|
|
resp = client.get("/api/health")
|
|
assert resp.headers["X-Content-Type-Options"] == "nosniff"
|
|
assert resp.headers["X-Frame-Options"] == "DENY"
|
|
assert "default-src 'self'" in resp.headers["Content-Security-Policy"]
|
|
assert "frame-ancestors 'none'" in resp.headers["Content-Security-Policy"]
|
|
|
|
|
|
def test_session_cookie_is_hardened(client, session):
|
|
make_user(session, username="u", role=UserRole.approver)
|
|
resp = client.post("/api/auth/login", json={"login": "u", "password": "password123"})
|
|
assert resp.status_code == 200
|
|
set_cookie = resp.headers["set-cookie"].lower()
|
|
assert "httponly" in set_cookie
|
|
assert "samesite=lax" in set_cookie
|
|
assert "max-age=" in set_cookie # bounded lifetime, not the 14-day default
|
|
|
|
|
|
def test_download_is_no_store(client, session, tmp_path, monkeypatch):
|
|
"""Document downloads must not be cached on shared machines."""
|
|
from ten31portal import storage
|
|
from ten31portal.models import Document, DocumentCategory, Entity, EntityType
|
|
|
|
monkeypatch.setattr(storage, "DOCS_DIR", str(tmp_path))
|
|
(tmp_path / "file.pdf").write_bytes(b"%PDF-1.4 test")
|
|
ent = Entity(name="F", type=EntityType.fund)
|
|
session.add(ent)
|
|
session.commit()
|
|
doc = Document(entity_id=ent.id, category=DocumentCategory.k1, title="K1",
|
|
original_filename="k1.pdf", content_type="application/pdf",
|
|
size_bytes=13, storage_path="file.pdf", uploaded_by=None)
|
|
session.add(doc)
|
|
session.commit()
|
|
make_user(session, username="staff", role=UserRole.approver)
|
|
_login(client, "staff")
|
|
|
|
resp = client.get(f"/api/documents/{doc.id}/download")
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.headers["Cache-Control"] == "no-store"
|
|
|
|
|
|
# --- P2-9: import preview must not leak cross-fund investors ------------------------------
|
|
|
|
def test_external_admin_preview_requires_a_fund(client, session, approver):
|
|
_two_funds(session)
|
|
_login(client, "fundadmin")
|
|
# No entity_id: refused before any name-matching can leak which accounts exist.
|
|
resp = client.post(
|
|
"/api/import/capital-accounts/preview",
|
|
files={"file": ("x.xlsx", b"not-a-real-workbook", "application/octet-stream")},
|
|
)
|
|
assert resp.status_code == 400, resp.text
|