476 lines
18 KiB
Python
476 lines
18 KiB
Python
"""The module handoff: what the platform tells another system about a user.
|
|
|
|
This is the highest-consequence surface in the application and had the least
|
|
coverage. A handoff is a signed statement — "this is who they are, this is what
|
|
they may do" — carried by the user's own browser to a system that has no other
|
|
way to check. Everything the signature does not cover is editable in dev tools,
|
|
and everything the platform forgets to check, the receiving module trusts.
|
|
|
|
Findings S-3 and S-4 were both here. These tests hold the fixes and cover the
|
|
paths that were never exercised at all.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from datetime import date, timedelta
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.core.tenant_context import unscoped
|
|
from app.services.auth.sso_service import SIGNATURE_VERSION, SSOService
|
|
|
|
from .conftest import requires_db, utc_today
|
|
|
|
pytestmark = requires_db
|
|
|
|
SECRET = "test-hmac-secret"
|
|
|
|
|
|
@pytest.fixture
|
|
def handoff(db, tenant_factory, plan_factory, role_factory, user_factory,
|
|
module_factory, environment_factory, tenant_module_factory,
|
|
module_access_factory):
|
|
"""A workspace entitled to a module, with one user who may use it.
|
|
|
|
`report.view` is granted by both the role and the plan, so it survives the
|
|
intersection. `report.delete` is granted by the plan only — it must not
|
|
appear, or the plan is granting permissions rather than bounding them.
|
|
"""
|
|
from types import SimpleNamespace
|
|
|
|
plan = plan_factory()
|
|
tenant = tenant_factory(plan_id=plan.id, end_date=utc_today() + timedelta(days=30))
|
|
role = role_factory(tenant=tenant, name="Analyst")
|
|
user = user_factory(tenant=tenant, role=role)
|
|
module = module_factory()
|
|
env = environment_factory(module, secret=SECRET)
|
|
link = tenant_module_factory(tenant, module)
|
|
|
|
module_access_factory(module, "report.view", role=role, plan=plan)
|
|
module_access_factory(module, "report.delete", plan=plan)
|
|
module_access_factory(module, "report.export", role=role)
|
|
|
|
with unscoped():
|
|
db.refresh(user)
|
|
db.refresh(role)
|
|
|
|
return SimpleNamespace(
|
|
plan=plan, tenant=tenant, role=role, user=user,
|
|
module=module, env=env, link=link,
|
|
)
|
|
|
|
|
|
def _sign(payload: dict, secret: str = SECRET) -> str:
|
|
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
|
|
return hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).hexdigest()
|
|
|
|
|
|
def test_the_whole_payload_is_covered_by_the_signature(db, handoff):
|
|
"""S-3. The signature covered four fields of an eleven-field payload.
|
|
|
|
`permissions`, `role`, `tenant_name` and the whole `subscription` object were
|
|
outside it, and the courier is the user's own browser.
|
|
"""
|
|
with unscoped():
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
|
|
assert result["headers"]["X-Signature"] == _sign(result["payload"])
|
|
assert result["headers"]["X-Signature-Version"] == SIGNATURE_VERSION
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"field, tampered",
|
|
[
|
|
("permissions", ["report.view", "report.delete", "admin.everything"]),
|
|
("role", "Superadmin"),
|
|
("tenant_id", "00000000-0000-0000-0000-000000000001"),
|
|
("email", "someone.else@example.com"),
|
|
("module_id", "another-module"),
|
|
("environment", "staging"),
|
|
("expires_at", 99999999999999),
|
|
],
|
|
)
|
|
def test_editing_any_field_invalidates_the_signature(db, handoff, field, tampered):
|
|
"""Each of these was individually forgeable under the old scheme."""
|
|
with unscoped():
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
|
|
forged = dict(result["payload"])
|
|
forged[field] = tampered
|
|
assert _sign(forged) != result["headers"]["X-Signature"]
|
|
|
|
|
|
def test_the_recipient_is_named_inside_the_signature(db, handoff):
|
|
"""So a payload minted for one module cannot be presented to another."""
|
|
with unscoped():
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
|
|
assert result["payload"]["module_id"] == handoff.module.module_id
|
|
assert result["payload"]["environment"] == handoff.env.slug
|
|
|
|
|
|
def test_the_payload_carries_replay_controls(db, handoff):
|
|
"""The platform never sees the POST, so the receiver is the only party that
|
|
can reject a replay. Both fields have to be inside the signature to be worth
|
|
anything."""
|
|
with unscoped():
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
|
|
payload = result["payload"]
|
|
assert payload["nonce"]
|
|
assert payload["expires_at"] > payload["issued_at"]
|
|
assert (payload["expires_at"] - payload["issued_at"]) <= 300_000
|
|
|
|
|
|
def test_two_handoffs_do_not_share_a_nonce(db, handoff):
|
|
with unscoped():
|
|
first = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
second = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert first["payload"]["nonce"] != second["payload"]["nonce"]
|
|
|
|
|
|
def test_the_app_id_header_names_the_module(db, handoff):
|
|
"""S-4's neighbour: the key was declared twice in one dict literal, so the
|
|
first value never shipped and receivers saw a different id than documented."""
|
|
with unscoped():
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert result["headers"]["X-App-Id"] == handoff.module.module_id
|
|
|
|
|
|
def test_permissions_are_the_role_bounded_by_the_plan(db, handoff):
|
|
"""S-2, in its second location.
|
|
|
|
The handoff read the plan's module codes and fell back to the role's only
|
|
when the plan set was empty — which handed every user every module permission
|
|
the plan carried, whatever their role.
|
|
"""
|
|
with unscoped():
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
|
|
assert result["payload"]["permissions"] == ["report.view"]
|
|
|
|
|
|
def test_a_lapsed_subscription_cannot_hand_off(db, handoff, tenant_factory):
|
|
"""A handoff from a dead workspace is a statement the platform should not make."""
|
|
with unscoped():
|
|
handoff.tenant.end_date = utc_today() - timedelta(days=1)
|
|
handoff.tenant.status = "EXPIRED"
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_a_workspace_without_the_module_cannot_hand_off(db, handoff):
|
|
"""The permissive path was the one in use.
|
|
|
|
`generate_grant` raised 403 when the workspace had no active `TenantModule`;
|
|
this path — the one the route actually calls — only read the row to pick an
|
|
environment slug and proceeded regardless. Two entry points to one trust
|
|
decision, and the wrong one was live.
|
|
"""
|
|
with unscoped():
|
|
handoff.link.is_active = False
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_a_disabled_module_cannot_hand_off(db, handoff):
|
|
with unscoped():
|
|
handoff.module.status = "disabled"
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_a_module_with_no_environment_fails_loudly(db, handoff, module_factory):
|
|
"""Rather than handing back a URL built from None."""
|
|
module = module_factory()
|
|
with unscoped(), pytest.raises(HTTPException) as exc:
|
|
SSOService.generate_signed_payload(
|
|
db, handoff.user.id, module.module_id, handoff.tenant.id
|
|
)
|
|
assert exc.value.status_code in (403, 404)
|
|
|
|
|
|
def test_the_assigned_environment_is_the_one_used(db, handoff, environment_factory):
|
|
"""A workspace pinned to staging must not be handed the production secret."""
|
|
staging = environment_factory(handoff.module, slug="staging", secret="staging-only")
|
|
with unscoped():
|
|
handoff.link.assigned_environment_slug = "staging"
|
|
db.flush()
|
|
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
|
|
assert result["payload"]["environment"] == "staging"
|
|
assert result["target_url"].startswith(staging.backend_base_url)
|
|
assert result["headers"]["X-Signature"] == _sign(result["payload"], "staging-only")
|
|
assert result["headers"]["X-Signature"] != _sign(result["payload"], SECRET)
|
|
|
|
|
|
def test_the_trust_secret_never_appears_in_the_handoff(db, handoff):
|
|
"""The signature proves possession of the secret; shipping it would defeat
|
|
the point of signing at all."""
|
|
with unscoped():
|
|
result = SSOService.generate_signed_payload(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert SECRET not in json.dumps(result)
|
|
|
|
|
|
def test_a_grant_is_good_exactly_once(db, handoff, fake_redis, module_signing_key):
|
|
"""Redeeming twice is the cheapest attack on a code that travels in a URL,
|
|
where it lands in browser history, referrers and server logs."""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
first = SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
assert first["access_token"]
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_an_unknown_grant_is_refused(db, handoff, fake_redis):
|
|
with unscoped(), pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(db, "not-a-real-code", handoff.module.module_id, "prod")
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_a_grant_cannot_be_redeemed_by_another_module(db, handoff, fake_redis,
|
|
module_factory, environment_factory):
|
|
"""Otherwise any module that saw a code could impersonate the user at another."""
|
|
other = module_factory()
|
|
environment_factory(other, secret="other-secret")
|
|
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(db, grant["grant_code"], other.module_id, "prod")
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_a_grant_cannot_be_redeemed_against_another_environment(db, handoff, fake_redis,
|
|
environment_factory):
|
|
"""A production grant redeemed at staging would cross a trust boundary."""
|
|
environment_factory(handoff.module, slug="staging", secret="staging-only")
|
|
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "staging"
|
|
)
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_a_grant_is_bound_to_the_workspace_it_was_minted_for(db, handoff, fake_redis,
|
|
tenant_factory, user_factory):
|
|
"""If the user moved workspace between minting and redeeming, the grant's
|
|
claim about which workspace they belong to is stale."""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
other = tenant_factory()
|
|
handoff.user.tenant_id = other.id
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_a_grant_needs_the_workspace_to_hold_the_module(db, handoff, fake_redis):
|
|
with unscoped():
|
|
handoff.link.is_active = False
|
|
db.flush()
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_the_exchanged_token_carries_the_bounded_permissions(
|
|
db, handoff, fake_redis, module_signing_key
|
|
):
|
|
"""The permissions are re-derived at exchange, not carried in the grant —
|
|
so a plan change between minting and redeeming is honoured."""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
result = SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
|
|
import jwt
|
|
|
|
claims = jwt.decode(
|
|
result["access_token"],
|
|
module_signing_key,
|
|
algorithms=["RS256"],
|
|
audience=handoff.module.module_id,
|
|
)
|
|
assert claims["permissions"] == ["report.view"]
|
|
assert claims["module_id"] == handoff.module.module_id
|
|
assert claims["environment"] == "prod"
|
|
|
|
|
|
def test_redis_being_down_is_a_refusal_not_a_bypass(db, handoff, monkeypatch):
|
|
"""Fail closed. A grant store that is unreachable must not mean "let them in"."""
|
|
from app.core import redis as redis_module
|
|
|
|
monkeypatch.setattr(redis_module.sync_redis_client, "_redis", None, raising=False)
|
|
|
|
with unscoped(), pytest.raises(HTTPException) as exc:
|
|
SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
assert exc.value.status_code == 503
|
|
|
|
|
|
def test_a_disabled_account_cannot_redeem_a_grant(db, handoff, fake_redis,
|
|
module_signing_key):
|
|
"""Login refuses a disabled account and so does the middleware. This path
|
|
never consulted `status` at all, so disabling someone left them a working
|
|
route into every module for as long as their grant lived."""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
handoff.user.status = "inactive"
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
assert exc.value.status_code == 401
|
|
|
|
|
|
def test_a_lapsed_workspace_cannot_redeem_a_grant(db, handoff, fake_redis,
|
|
module_signing_key):
|
|
"""The signed-payload path refuses a dead subscription; this one did not.
|
|
|
|
It produced a token with an empty permission list, which is not the same as
|
|
a refusal: a module that treats "authenticated" as sufficient — and modules
|
|
do — admitted the user anyway.
|
|
"""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
handoff.tenant.end_date = utc_today() - timedelta(days=1)
|
|
handoff.tenant.status = "EXPIRED"
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_revoking_a_module_invalidates_grants_in_flight(db, handoff, fake_redis,
|
|
module_signing_key):
|
|
"""Entitlement was checked when the grant was minted and never again."""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
handoff.link.is_active = False
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
assert exc.value.status_code == 403
|
|
|
|
|
|
def test_disabling_a_module_invalidates_grants_in_flight(db, handoff, fake_redis,
|
|
module_signing_key):
|
|
"""Taking a module offline should stop admitting people to it."""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
handoff.module.status = "disabled"
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
assert exc.value.status_code in (401, 403)
|
|
|
|
|
|
def test_a_grant_is_still_consumed_when_the_exchange_is_refused(db, handoff, fake_redis):
|
|
"""The code is deleted before any of these checks run, and must stay deleted.
|
|
|
|
Otherwise a refusal hands back a code that can be retried the moment the
|
|
refusing condition clears — and turns each check into a way to probe the
|
|
workspace's state.
|
|
"""
|
|
with unscoped():
|
|
grant = SSOService.generate_grant(
|
|
db, handoff.user.id, handoff.module.module_id, handoff.tenant.id
|
|
)
|
|
handoff.user.status = "inactive"
|
|
db.flush()
|
|
|
|
with pytest.raises(HTTPException):
|
|
SSOService.exchange_grant(
|
|
db, grant["grant_code"], handoff.module.module_id, "prod"
|
|
)
|
|
|
|
assert fake_redis.get(f"sso_grant:{grant['grant_code']}") is None
|