0.2.46: security hardening (external-fund scope, role guards, forced password rotation, headers)
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.
This commit is contained in:
Jonathan Kirkwood
@@ -0,0 +1,256 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user