"""FlowDeck — v6.7.0 : SSO/SAML 2.0 + OIDC (enterprise auth).
Covers the admin config API (validation, secrets, disable), the SP metadata
endpoint, a full SP-initiated SAML login against a mock IdP (real signatures
produced by python3-saml's own ``add_sign``), every rejection path of the
security table of the design doc (signature, expiry, audience, destination,
issuer, InResponseTo, replay, CSRF relay), auto-provisioning + attribute
merge + group → workspace role mapping, the SSO-only restriction on local
auth, the audit trail, the rate limiter, and the OIDC flow with a mocked
provider (discovery / token / JWKS + signed ID token).
"""
from __future__ import annotations
import base64
import datetime as dt
import secrets as pysecrets
from urllib.parse import parse_qs, urlparse
import pytest
from conftest import anon
# ── Mock IdP constants ─────────────────────────────────────────────────────
IDP_ENTITY = "https://idp.corp.test/saml/metadata"
IDP_SSO = "https://idp.corp.test/saml/sso"
IDP_SLO = "https://idp.corp.test/saml/slo"
SAML_NS_ASSERTION = "urn:oasis:names:tc:SAML:2.0:assertion"
SAML_NS_PROTOCOL = "urn:oasis:names:tc:SAML:2.0:protocol"
# ── Fixtures / helpers ─────────────────────────────────────────────────────
@pytest.fixture(scope="session")
def idp_keypair() -> tuple[str, str]:
"""RSA key + self-signed certificate used by the mock IdP (once per run)."""
import datetime
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "idp.corp.test")])
now = datetime.datetime.now(datetime.UTC)
cert = (
x509.CertificateBuilder()
.subject_name(name)
.issuer_name(name)
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now - datetime.timedelta(days=1))
.not_valid_after(now + datetime.timedelta(days=3650))
.sign(key, hashes.SHA256())
)
key_pem = key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
).decode()
cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
return key_pem, cert_pem
_ADMIN_TOKENS: dict[int, str] = {}
def _clear_session_cookie(client) -> None:
"""Drop EVERY flowdeck_session cookie in the jar (client- and server-set).
httpx keys cookies by (name, domain, path): a cookie we set by hand has
domain='' while a server Set-Cookie lands with domain='testserver', so
both coexist and ``jar.get(name)`` then raises CookieConflict. Clear
before a request that will set a fresh session.
"""
client.cookies.delete("flowdeck_session")
def _be_admin(client) -> None:
"""Make sure the ONLY session cookie in the jar is the admin one."""
_clear_session_cookie(client)
client.cookies.set("flowdeck_session", _ADMIN_TOKENS[id(client)])
def _admin_session(client) -> tuple[int, str]:
"""Create an admin user and put its session + CSRF cookies on the client."""
from app.auth.session import SessionManager
from app.db import get_conn
login = f"ssoadmin_{pysecrets.token_hex(3)}"
with get_conn() as conn:
conn.execute(
"INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'SSO Admin', ?, 1)",
(login, f"{login}@test.dev"),
)
row = conn.execute("SELECT * FROM users WHERE login=?", (login,)).fetchone()
conn.commit()
user = dict(row)
token = SessionManager.create_session(user)
_ADMIN_TOKENS[id(client)] = token
_clear_session_cookie(client)
client.cookies.set("flowdeck_session", token)
client.cookies.set("csrf_token", "csrf-sso-test-token")
return user["id"], login
def _save_config(client, payload: dict) -> dict:
r = client.post(
"/api/v2/sso/config", json=payload, headers={"X-CSRF-Token": "csrf-sso-test-token"}
)
assert r.status_code == 200, r.text
return r.json()
def _saml_payload(cert_pem: str, **overrides) -> dict:
payload = {
"provider_type": "saml",
"name": "Corp SSO",
"entity_id": IDP_ENTITY,
"sso_url": IDP_SSO,
"slo_url": IDP_SLO,
"x509_certificate": cert_pem,
"attribute_mapping": {},
"groups_mapping": [],
"auto_provision": 1,
"sso_only": 0,
"sign_requests": 0,
"default_workspace_id": None,
}
payload.update(overrides)
return payload
def _configure_saml(client, cert_pem: str, **overrides) -> dict:
_admin_session(client)
return _save_config(client, _saml_payload(cert_pem, **overrides))
def _start_saml(client, next_path: str = "/workspaces") -> tuple[str, str, str]:
"""Kick the SP-initiated flow → (authn_request_id, csrf relay, location)."""
r = client.get(f"/auth/saml/login?next={next_path}", follow_redirects=False)
assert r.status_code == 302, r.text
location = r.headers["location"]
assert location.startswith(IDP_SSO), location
query = parse_qs(urlparse(location).query)
assert query.get("SAMLRequest"), "AuthnRequest missing from the redirect"
relay = query["RelayState"][0]
rid, _, csrf = relay.partition(".")
assert rid and csrf
return rid, csrf, location
def _sp_endpoints(client) -> tuple[str, str]:
"""(sp_entity_id, acs_url) as advertised by our own metadata.
Careful: in SP metadata the SingleLogoutService comes BEFORE the
AssertionConsumerService, so grabbing the first ``Location=`` would
return the SLO URL — read the ACS element explicitly.
"""
r = client.get("/auth/saml/metadata")
assert r.status_code == 200, r.text
xml = r.text
entity = xml.split('entityID="', 1)[1].split('"', 1)[0]
acs_block = xml.split("AssertionConsumerService", 1)[1]
acs = acs_block.split('Location="', 1)[1].split('"', 1)[0]
return entity, acs
def _iso(moment: dt.datetime) -> str:
return moment.strftime("%Y-%m-%dT%H:%M:%S.000Z")
def _attributes_xml(attributes: dict[str, list[str] | str]) -> str:
if not attributes:
return ""
rows = []
for name, values in attributes.items():
if isinstance(values, str):
values = [values]
vals = "".join(f"{v}" for v in values)
rows.append(
f'{vals}'
)
return "" + "".join(rows) + ""
def build_saml_response(
key_pem: str,
cert_pem: str,
*,
sp_entity: str,
acs_url: str,
in_response_to: str,
name_id: str = "alice@corp.test",
attributes: dict | None = None,
audience: str | None = None,
destination: str | None = None,
issuer: str | None = None,
not_on_or_after: dt.datetime | None = None,
status_ok: bool = True,
) -> str:
"""Build + sign (assertion) a SAMLResponse like a real IdP would, base64."""
from onelogin.saml2.utils import OneLogin_Saml2_Utils
issuer = issuer or IDP_ENTITY
audience = audience or sp_entity
destination = destination or acs_url
now = dt.datetime.now(dt.UTC)
not_on_or_after = not_on_or_after or (now + dt.timedelta(minutes=5))
status = (
""
if status_ok
else ""
)
assertion = f"""
{issuer}
{name_id}
{audience}
urn:oasis:names:tc:SAML:2.0:ac:classes:Password
{_attributes_xml(attributes if attributes else {"email": [name_id]})}
"""
signed = OneLogin_Saml2_Utils.add_sign(assertion, key_pem, cert_pem)
if isinstance(signed, bytes):
signed = signed.decode()
signed_assertion = signed
if signed_assertion.startswith("", 1)[1].strip()
response = f"""
{issuer}
{status}
{signed_assertion}
"""
return base64.b64encode(response.encode()).decode()
def _post_response(client, b64_response: str, relay: str):
return client.post(
"/auth/saml/callback",
data={"SAMLResponse": b64_response, "RelayState": relay},
follow_redirects=False,
)
def _history(client, limit=50):
r = client.get(f"/api/v2/sso/history?limit={limit}")
assert r.status_code == 200, r.text
return r.json()["history"]
# ── Migration ──────────────────────────────────────────────────────────────
def test_sso_tables_exist(client):
"""Migration 23 creates sso_config / sso_login_history / sso_requests."""
from app.db import get_conn
with get_conn() as conn:
tables = {
r["name"]
for r in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
assert {"sso_config", "sso_login_history", "sso_requests"} <= tables
from app.migrations import current_version
assert current_version(conn) >= 23
# ── Admin config API ───────────────────────────────────────────────────────
def test_providers_endpoint_empty_without_config(client):
r = client.get("/api/v2/sso/providers")
assert r.status_code == 200
assert r.json() == {"providers": [], "sso_only": False}
def test_config_requires_admin(client):
anon(client)
assert client.get("/api/v2/sso/config").status_code == 401
_admin_session(client)
# demote to plain user → 403
from app.db import get_conn
with get_conn() as conn:
conn.execute("UPDATE users SET is_admin=0 WHERE login LIKE 'ssoadmin_%'")
conn.commit()
assert client.get("/api/v2/sso/config").status_code == 403
def test_config_rejects_invalid_payloads(client, idp_keypair):
_, cert_pem = idp_keypair
_admin_session(client)
headers = {"X-CSRF-Token": "csrf-sso-test-token"}
missing_cert = _saml_payload(cert_pem)
missing_cert["x509_certificate"] = ""
assert client.post("/api/v2/sso/config", json=missing_cert, headers=headers).status_code == 400
not_pem = _saml_payload(cert_pem)
not_pem["x509_certificate"] = "not-a-certificate"
assert client.post("/api/v2/sso/config", json=not_pem, headers=headers).status_code == 400
no_url = _saml_payload(cert_pem)
no_url["sso_url"] = ""
assert client.post("/api/v2/sso/config", json=no_url, headers=headers).status_code == 400
bad_oidc = {"provider_type": "oidc", "issuer_url": "https://issuer.test"}
assert client.post("/api/v2/sso/config", json=bad_oidc, headers=headers).status_code == 400
bad_mapping = _saml_payload(cert_pem, attribute_mapping="[]")
assert client.post("/api/v2/sso/config", json=bad_mapping, headers=headers).status_code == 400
def test_config_roundtrip_and_secrets_never_leak(client, idp_keypair):
key_pem, cert_pem = idp_keypair
saved = _configure_saml(client, cert_pem)
assert saved["configured"] is True
assert saved["provider_type"] == "saml"
assert saved["provisioned_users"] == 0
# The PEM IdP certificate is public data, the SP private key is not.
assert saved["x509_certificate"].strip().startswith("-----BEGIN CERTIFICATE-----")
assert "private" not in " ".join(saved.keys())
assert "sp_private_key" not in saved
r = client.get("/api/v2/sso/config")
body = r.text
assert "BEGIN PRIVATE KEY" not in body
assert key_pem.splitlines()[1] not in body
# Public providers endpoint advertises the login button.
providers = client.get("/api/v2/sso/providers").json()
assert providers["providers"][0]["type"] == "saml"
assert providers["providers"][0]["login_url"].startswith("/auth/saml/login")
assert providers["sso_only"] is False
def test_config_secret_roundtrip_keeps_existing_value(client, idp_keypair):
"""A blank client_secret on save must not wipe the stored one."""
_, cert_pem = idp_keypair
_admin_session(client)
headers = {"X-CSRF-Token": "csrf-sso-test-token"}
oidc = {
"provider_type": "oidc",
"issuer_url": "https://issuer.test",
"client_id": "flowdeck-client",
"client_secret": "super-secret",
}
assert client.post("/api/v2/sso/config", json=oidc, headers=headers).status_code == 200
from app.services import sso_provisioning as sso
cfg = sso.get_sso_config()
assert sso.client_secret_value(cfg) == "super-secret"
assert cfg["client_secret"] != "super-secret" # encrypted at rest
# Save again with a blank secret → value preserved.
oidc_blank = dict(oidc, client_secret="")
assert client.post("/api/v2/sso/config", json=oidc_blank, headers=headers).status_code == 200
cfg = sso.get_sso_config()
assert sso.client_secret_value(cfg) == "super-secret"
def test_config_disable_turns_sso_off(client, idp_keypair):
_, cert_pem = idp_keypair
_configure_saml(client, cert_pem)
assert client.get("/api/v2/sso/providers").json()["providers"]
r = client.delete(
"/api/v2/sso/config", headers={"X-CSRF-Token": "csrf-sso-test-token"}
)
assert r.status_code == 200, r.text
# SSO off → no button, local login still available.
assert client.get("/api/v2/sso/providers").json() == {
"providers": [],
"sso_only": False,
}
assert client.get("/auth/saml/login", follow_redirects=False).status_code == 404
def test_workspaces_and_sync_endpoints(client, idp_keypair):
_, cert_pem = idp_keypair
uid, _ = _admin_session(client)
_save_config(client, _saml_payload(cert_pem))
from app.db import get_conn
with get_conn() as conn:
ws = conn.execute(
"INSERT INTO workspaces (name, owner_id) VALUES ('Corp', ?)", (uid,)
).lastrowid
conn.execute(
"INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?, ?, 'editor')",
(ws, uid),
)
conn.execute(
"INSERT INTO users (login, full_name, email, auth_method) VALUES "
"('sso-user', 'SSO User', 'sso-user@corp.test', 'saml')"
)
conn.commit()
data = client.get("/api/v2/sso/workspaces").json()
assert any(w["id"] == ws for w in data["workspaces"])
assert data["provisioned_users"] == 1
r = client.post(
"/api/v2/sso/sync", headers={"X-CSRF-Token": "csrf-sso-test-token"}
)
assert r.status_code == 200, r.text
assert r.json()["status"] == "ok"
assert r.json()["users"] == 1
# ── SP metadata ────────────────────────────────────────────────────────────
def test_metadata_is_valid_sp_descriptor(client, idp_keypair):
_, cert_pem = idp_keypair
_configure_saml(client, cert_pem)
r = client.get("/auth/saml/metadata")
assert r.status_code == 200
assert "application/samlmetadata+xml" in r.headers["content-type"]
xml = r.text
assert "").decode(), "RelayState": "x.y"},
follow_redirects=False,
)
assert r.status_code == 403
# ── Group mapping ──────────────────────────────────────────────────────────
def test_group_mapping_sets_workspace_role(client, idp_keypair):
key_pem, cert_pem = idp_keypair
uid, _ = _admin_session(client)
_save_config(
client,
_saml_payload(
cert_pem,
groups_mapping=[
{"sso_group": "FlowDeck Admins", "workspace_role": "admin"},
{"sso_group": "FlowDeck Members", "workspace_role": "viewer"},
],
),
)
from app.db import get_conn
with get_conn() as conn:
default_ws = conn.execute(
"INSERT INTO workspaces (name, owner_id) VALUES ('Default WS', ?)", (uid,)
).lastrowid
mapped_ws = conn.execute(
"INSERT INTO workspaces (name, owner_id) VALUES ('Mapped WS', ?)", (uid,)
).lastrowid
conn.commit()
# Re-save so default_workspace_id / mapping point at real workspaces.
_save_config(
client,
_saml_payload(
cert_pem,
default_workspace_id=default_ws,
groups_mapping=[
{"sso_group": "FlowDeck Admins", "workspace_role": "admin",
"workspace_id": mapped_ws},
],
),
)
sp_entity, acs = _sp_endpoints(client)
rid, csrf, _ = _start_saml(client)
signed = build_saml_response(
key_pem, cert_pem, sp_entity=sp_entity, acs_url=acs, in_response_to=rid,
name_id="bob@corp.test",
attributes={"email": "bob@corp.test", "groups": ["FlowDeck Admins", "Other Group"]},
)
r = _post_response(client, signed, f"{rid}.{csrf}")
assert r.status_code == 302, r.text
with get_conn() as conn:
user = conn.execute(
"SELECT id FROM users WHERE email='bob@corp.test'"
).fetchone()
mapped = conn.execute(
"SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
(mapped_ws, user["id"]),
).fetchone()
default_member = conn.execute(
"SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
(default_ws, user["id"]),
).fetchone()
assert mapped is not None and mapped["role"] == "admin"
assert default_member is not None # default workspace membership
# ── SSO-only restriction (design §7.1) ─────────────────────────────────────
def test_sso_only_blocks_local_login_and_register(client, idp_keypair):
_, cert_pem = idp_keypair
_configure_saml(client, cert_pem, sso_only=1)
from app.db import get_conn
from app.password_utils import hash_password
with get_conn() as conn:
conn.execute(
"INSERT INTO users (login, full_name, email, password_hash) VALUES "
"('localguy', 'Local Guy', 'localguy@t.dev', ?)",
(hash_password("secret123"),),
)
# An admin with a local password — the allowed back door (§7.1).
conn.execute(
"INSERT INTO users (login, full_name, email, is_admin, password_hash) VALUES "
"('keepadmin', 'Keep Admin', 'keepadmin@t.dev', 1, ?)",
(hash_password("adminpass1"),),
)
conn.commit()
r = client.post(
"/auth/local-login", json={"email": "localguy", "password": "secret123"}
)
assert r.status_code == 403
assert "SSO" in r.json()["error"]
r2 = client.post(
"/auth/register", json={"email": "new@t.dev", "password": "secret123"}
)
assert r2.status_code == 403
# Admins keep their local door open (design: « l'admin garde le sien »).
_clear_session_cookie(client)
admin_login = client.post(
"/auth/local-login", json={"email": "keepadmin", "password": "adminpass1"}
)
assert admin_login.status_code == 200, admin_login.text
# The login page tells the UI to hide the local form.
page = client.get("/auth/login?provider=local")
assert page.status_code == 200
assert 'id="sso-section"' in page.text
assert "loadSsoProviders" in page.text
def test_local_login_still_works_when_sso_is_optional(client, idp_keypair):
_, cert_pem = idp_keypair
_configure_saml(client, cert_pem, sso_only=0)
from app.db import get_conn
from app.password_utils import hash_password
with get_conn() as conn:
conn.execute(
"INSERT INTO users (login, full_name, email, password_hash) VALUES "
"('mixeduser', 'Mixed', 'mixed@t.dev', ?)",
(hash_password("secret123"),),
)
conn.commit()
_clear_session_cookie(client) # local login will set its own session
r = client.post("/auth/local-login", json={"email": "mixeduser", "password": "secret123"})
assert r.status_code == 200, r.text
assert client.cookies.get("flowdeck_session")
# ── Rate limiting (design §5.2) ────────────────────────────────────────────
def test_saml_login_rate_limited(client, idp_keypair):
_, cert_pem = idp_keypair
_configure_saml(client, cert_pem)
from app.config import settings
from app.routers import sso as sso_router
previous = settings.rate_limit_enabled
settings.rate_limit_enabled = True
sso_router._rate_store.clear()
try:
statuses = [
client.get("/auth/saml/login", follow_redirects=False).status_code
for _ in range(7)
]
finally:
settings.rate_limit_enabled = previous
sso_router._rate_store.clear()
assert statuses[:5] == [302] * 5
assert statuses[5:] == [429, 429]
# ── SLO ────────────────────────────────────────────────────────────────────
def test_saml_logout_sp_initiated_builds_logout_request(client, idp_keypair):
key_pem, cert_pem = idp_keypair
_configure_saml(client, cert_pem)
sp_entity, acs = _sp_endpoints(client)
rid, csrf, _ = _start_saml(client)
signed = build_saml_response(
key_pem, cert_pem, sp_entity=sp_entity, acs_url=acs, in_response_to=rid,
name_id="slo@corp.test", attributes={"email": "slo@corp.test"},
)
_clear_session_cookie(client) # only the SSO session must be in the jar
assert _post_response(client, signed, f"{rid}.{csrf}").status_code == 302
from app.auth.session import SessionManager
sso_sid = SessionManager.session_id(client.cookies.get("flowdeck_session") or "")
assert sso_sid # the SSO session we are about to kill
# /auth/logout hands SSO sessions over to the SLO route…
r = client.get("/auth/logout", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"].startswith("/auth/saml/logout")
# …which revokes the session locally and redirects to the IdP SLO URL.
r2 = client.get("/auth/saml/logout?next=/auth/login?provider=local",
follow_redirects=False)
assert r2.status_code == 302
assert r2.headers["location"].startswith(IDP_SLO)
assert "SAMLRequest" in r2.headers["location"]
assert r2.headers.get("set-cookie", "").find("flowdeck_session=;") >= 0 or \
"flowdeck_session=\"\"" in r2.headers.get("set-cookie", "")
# The SSO session is revoked server-side (the admin's stays alive).
from app.db import get_conn
with get_conn() as conn:
row = conn.execute(
"SELECT revoked FROM user_sessions WHERE id=?", (sso_sid,)
).fetchone()
assert row and row["revoked"] == 1
cookie = client.cookies.get("flowdeck_session")
assert not cookie or SessionManager.decode_session(cookie) is None
# ── OIDC ───────────────────────────────────────────────────────────────────
def _oidc_payload(**overrides) -> dict:
payload = {
"provider_type": "oidc",
"name": "Corp OIDC",
"issuer_url": "https://issuer.corp.test",
"client_id": "flowdeck-client",
"client_secret": "s3cr3t",
"scope": "openid profile email groups",
"attribute_mapping": {},
"groups_mapping": [],
"auto_provision": 1,
"sso_only": 0,
}
payload.update(overrides)
return payload
def _sign_id_token(key_pem: str, kid: str, payload: dict) -> str:
import warnings
from authlib.jose import jwt as jose_jwt
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
token = jose_jwt.encode({"alg": "RS256", "kid": kid}, payload, key_pem)
# IdP token endpoints return str JSON; authlib encodes to bytes.
return token.decode() if isinstance(token, bytes) else token
@pytest.fixture
def oidc_provider_env(monkeypatch, idp_keypair, client):
"""Save an OIDC config + mock discovery / token / userinfo / JWKS with a
signed ID token (no real network call can ever happen)."""
_admin_session(client)
_save_config(client, _oidc_payload())
key_pem, cert_pem = idp_keypair
import base64 as b64
import time as t
from cryptography.hazmat.primitives import serialization
from app.auth.providers import oidc_provider as op
priv = serialization.load_pem_private_key(key_pem.encode(), password=None)
pub = priv.public_key().public_numbers()
def b64u(i: int) -> str:
raw = i.to_bytes((i.bit_length() + 7) // 8, "big")
return b64.urlsafe_b64encode(raw).rstrip(b"=").decode()
jwk = {
"kty": "RSA", "kid": "oidc-test-key", "use": "sig", "alg": "RS256",
"n": b64u(pub.n), "e": b64u(pub.e),
}
doc = {
"issuer": "https://issuer.corp.test",
"authorization_endpoint": "https://issuer.corp.test/authorize",
"token_endpoint": "https://issuer.corp.test/token",
"userinfo_endpoint": "https://issuer.corp.test/userinfo",
"jwks_uri": "https://issuer.corp.test/jwks",
"end_session_endpoint": "https://issuer.corp.test/logout",
}
async def fake_discover(issuer_url: str) -> dict:
return doc
async def fake_exchange(doc_arg, **kwargs):
now = t.time()
id_token = _sign_id_token(key_pem, "oidc-test-key", {
"iss": "https://issuer.corp.test",
"aud": "flowdeck-client",
"sub": "oidc-sub-1",
"exp": int(now) + 300,
"iat": int(now),
"nonce": kwargs.get("_nonce") or _EXPECTED_NONCE["value"],
"email": "carol@corp.test",
"name": "Carol OIDC",
"groups": ["FlowDeck Admins"],
})
return {"id_token": id_token, "access_token": "at-123", "token_type": "Bearer"}
async def fake_userinfo(doc_arg, access_token: str) -> dict:
return {"preferred_username": "carol-oidc"}
async def fake_jwks(doc_arg) -> dict:
return {"keys": [jwk]}
monkeypatch.setattr(op, "discover", fake_discover)
monkeypatch.setattr(op, "exchange_code", fake_exchange)
monkeypatch.setattr(op, "fetch_userinfo", fake_userinfo)
import app.routers.sso as sso_router
monkeypatch.setattr(sso_router, "_fetch_jwks", fake_jwks)
return {"nonce": _EXPECTED_NONCE}
# The nonce the mocked token must carry is written here by the login step.
_EXPECTED_NONCE = {"value": ""}
def test_oidc_login_redirects_with_pkce(client, oidc_provider_env):
r = client.get("/auth/oidc/login?next=/workspaces", follow_redirects=False)
assert r.status_code == 302, r.text
location = r.headers["location"]
query = parse_qs(urlparse(location).query)
assert location.startswith("https://issuer.corp.test/authorize")
assert query["response_type"] == ["code"]
assert query["client_id"] == ["flowdeck-client"]
assert query["code_challenge_method"] == ["S256"]
assert query["scope"] == ["openid profile email groups"]
state = query["state"][0]
_EXPECTED_NONCE["value"] = query["nonce"][0]
from app.db import get_conn
with get_conn() as conn:
row = conn.execute(
"SELECT * FROM sso_requests WHERE id=? AND kind='oidc'", (state,)
).fetchone()
assert row is not None
assert row["relay_state"] == _EXPECTED_NONCE["value"]
assert row["code_verifier"]
def test_oidc_login_404_without_config(client):
assert client.get("/auth/oidc/login", follow_redirects=False).status_code == 404
def test_oidc_full_login_flow(client, oidc_provider_env):
r = client.get("/auth/oidc/login?next=/library", follow_redirects=False)
assert r.status_code == 302
query = parse_qs(urlparse(r.headers["location"]).query)
_EXPECTED_NONCE["value"] = query["nonce"][0]
state = query["state"][0]
_clear_session_cookie(client) # callback will set the SSO session
cb = client.get(
f"/auth/oidc/callback?code=abc123&state={state}", follow_redirects=False
)
assert cb.status_code == 302, cb.text
assert cb.headers["location"] == "/library"
assert client.cookies.get("flowdeck_session")
from app.db import get_conn
with get_conn() as conn:
user = conn.execute(
"SELECT * FROM users WHERE email='carol@corp.test'"
).fetchone()
hist = conn.execute(
"SELECT * FROM sso_login_history WHERE provider_type='oidc' AND success=1"
).fetchall()
assert user is not None
assert user["auth_method"] == "oidc"
assert user["full_name"] == "Carol OIDC"
assert len(hist) == 1
# Groups came through the attribute mapping.
assert "FlowDeck Admins" in hist[0]["sso_identifier"]
def test_oidc_rejects_unknown_state(client, oidc_provider_env):
admin_token = client.cookies.get("flowdeck_session")
cb = client.get(
"/auth/oidc/callback?code=abc&state=forged-state", follow_redirects=False
)
assert cb.status_code == 403
# Session untouched — still exactly the admin cookie we started with.
assert client.cookies.get("flowdeck_session") == admin_token
def test_oidc_rejects_replayed_state(client, oidc_provider_env):
r = client.get("/auth/oidc/login", follow_redirects=False)
query = parse_qs(urlparse(r.headers["location"]).query)
_EXPECTED_NONCE["value"] = query["nonce"][0]
state = query["state"][0]
first = client.get(
f"/auth/oidc/callback?code=abc&state={state}", follow_redirects=False
)
assert first.status_code == 302, first.text
client.cookies.delete("flowdeck_session")
second = client.get(
f"/auth/oidc/callback?code=abc&state={state}", follow_redirects=False
)
assert second.status_code == 403
assert client.cookies.get("flowdeck_session") is None
def test_oidc_rejects_wrong_nonce(client, monkeypatch, idp_keypair):
"""ID token nonce must match the one of the issued request (token swap)."""
import time as t
import app.routers.sso as sso_router
from app.auth.providers import oidc_provider as op
async def fake_discover(issuer_url):
return {
"issuer": "https://issuer.corp.test",
"authorization_endpoint": "https://issuer.corp.test/authorize",
"token_endpoint": "https://issuer.corp.test/token",
"jwks_uri": "https://issuer.corp.test/jwks",
}
async def fake_exchange(doc, **kwargs):
key_pem, _ = idp_keypair
id_token = _sign_id_token(key_pem, "oidc-test-key", {
"iss": "https://issuer.corp.test",
"aud": "flowdeck-client",
"sub": "attacker",
"exp": int(t.time()) + 300,
"iat": int(t.time()),
"nonce": "not-the-right-nonce",
"email": "attacker@corp.test",
})
return {"id_token": id_token, "access_token": "at"}
async def fake_jwks(doc):
import base64 as b64
from cryptography.hazmat.primitives import serialization
key_pem, _ = idp_keypair
priv = serialization.load_pem_private_key(key_pem.encode(), password=None)
pub = priv.public_key().public_numbers()
def b64u(i):
raw = i.to_bytes((i.bit_length() + 7) // 8, "big")
return b64.urlsafe_b64encode(raw).rstrip(b"=").decode()
return {"keys": [{"kty": "RSA", "kid": "oidc-test-key", "use": "sig",
"alg": "RS256", "n": b64u(pub.n), "e": b64u(pub.e)}]}
monkeypatch.setattr(op, "discover", fake_discover)
monkeypatch.setattr(op, "exchange_code", fake_exchange)
monkeypatch.setattr(op, "fetch_userinfo", lambda *a, **k: _empty_userinfo())
monkeypatch.setattr(sso_router, "_fetch_jwks", fake_jwks)
_admin_session(client)
_save_config(client, _oidc_payload())
r = client.get("/auth/oidc/login", follow_redirects=False)
assert r.status_code == 302, r.text
query = parse_qs(urlparse(r.headers["location"]).query)
state = query["state"][0]
cb = client.get(
f"/auth/oidc/callback?code=abc&state={state}", follow_redirects=False
)
assert cb.status_code == 403
assert "nonce" in cb.text.lower()
# Rejected token → no session issued, the admin cookie is untouched.
assert client.cookies.get("flowdeck_session") == _ADMIN_TOKENS[id(client)]
from app.db import get_conn
with get_conn() as conn:
user = conn.execute(
"SELECT 1 FROM users WHERE email='attacker@corp.test'"
).fetchone()
assert user is None
async def _empty_userinfo() -> dict:
return {}
# ── Env bootstrap fallback (design §3.3) ───────────────────────────────────
def test_env_config_fallback_when_table_empty(client, monkeypatch):
"""SSO_* env vars bootstrap a config when no admin ever saved one."""
from app.config import settings
monkeypatch.setattr(settings, "sso_provider", "saml")
monkeypatch.setattr(settings, "sso_entity_id", IDP_ENTITY)
monkeypatch.setattr(settings, "sso_sso_url", IDP_SSO)
monkeypatch.setattr(settings, "sso_x509_certificate", "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----")
from app.services import sso_provisioning as sso
cfg = sso.get_sso_config()
assert cfg is not None
assert cfg["_source"] == "env"
assert cfg["provider_type"] == "saml"
r = client.get("/api/v2/sso/providers")
assert r.json()["providers"][0]["type"] == "saml"
# The env config is visible but not admin-editable in the UI.
_admin_session(client)
assert client.get("/api/v2/sso/config").json()["source"] == "env"
# ── Audit trail ────────────────────────────────────────────────────────────
def test_history_endpoint_requires_admin_and_lists_attempts(client, idp_keypair):
key_pem, cert_pem = idp_keypair
_configure_saml(client, cert_pem)
sp_entity, acs = _sp_endpoints(client)
rid, csrf, _ = _start_saml(client)
# one failure (bad signature) + one success
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
rogue = rsa.generate_private_key(public_exponent=65537, key_size=2048)
rogue_pem = rogue.private_bytes(
serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
).decode()
bad = build_saml_response(
rogue_pem, cert_pem, sp_entity=sp_entity, acs_url=acs, in_response_to=rid,
)
assert _post_response(client, bad, f"{rid}.{csrf}").status_code == 403
rid2, csrf2, _ = _start_saml(client)
_clear_session_cookie(client) # success → server sets the SSO session
good = build_saml_response(
key_pem, cert_pem, sp_entity=sp_entity, acs_url=acs, in_response_to=rid2,
name_id="dave@corp.test", attributes={"email": "dave@corp.test"},
)
assert _post_response(client, good, f"{rid2}.{csrf2}").status_code == 302
_be_admin(client) # history endpoint needs the admin session back
history = _history(client)
assert len(history) >= 2
successes = [h for h in history if h["success"]]
failures = [h for h in history if not h["success"]]
assert len(successes) == 1
assert len(failures) == 1
assert failures[0]["error_message"]
# anonymous → 401
anon(client)
assert client.get("/api/v2/sso/history").status_code == 401
# ── PermissionManager extension (design §7.2) ─────────────────────────────
def test_permission_manager_sso_methods(client, idp_keypair):
"""§7.2 — is_sso_only_workspace / get_sso_roles / sync_sso_permissions."""
_, cert_pem = idp_keypair
uid, _ = _admin_session(client)
from app.db import get_conn
from app.services.permission_manager import PermissionManager
with get_conn() as conn:
ws = conn.execute(
"INSERT INTO workspaces (name, owner_id) VALUES ('Corp', ?)", (uid,)
).lastrowid
conn.execute(
"INSERT INTO users (login, full_name, email, auth_method) VALUES "
"('pm_local', 'PM Local', 'pm_local@t.dev', 'local')"
)
conn.execute(
"INSERT INTO users (login, full_name, email, auth_method) VALUES "
"('pm_sso', 'PM SSO', 'pm_sso@t.dev', 'saml')"
)
conn.commit()
pm_sso_id = conn.execute("SELECT id FROM users WHERE login='pm_sso'").fetchone()["id"]
pm_local_id = conn.execute("SELECT id FROM users WHERE login='pm_local'").fetchone()["id"]
pm_sso = PermissionManager(pm_sso_id)
pm_local = PermissionManager(pm_local_id)
# No SSO-only flag, no grants yet.
assert pm_sso.is_sso_only_workspace(ws) is False
assert pm_sso.get_sso_roles(workspace_id=ws) == []
# A local account never carries SSO roles, even once it becomes a member.
with get_conn() as conn:
conn.execute(
"INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?, ?, 'editor')",
(ws, pm_local_id),
)
conn.commit()
assert pm_local.get_sso_roles(workspace_id=ws) == []
# Configure the mapping, then re-apply it through the manager.
_save_config(
client,
_saml_payload(
cert_pem,
groups_mapping=[
{"sso_group": "FlowDeck Admins", "workspace_role": "admin", "workspace_id": ws}
],
),
)
touched = pm_sso.sync_sso_permissions(pm_sso_id, ["FlowDeck Admins"], ws)
assert touched == [ws]
assert pm_sso.role_in_workspace(ws) == "admin" # cache invalidated
assert pm_sso.get_sso_roles(workspace_id=ws) == ["admin"]
# Unknown group → nothing touched (and the role stays put).
assert pm_sso.sync_sso_permissions(pm_sso_id, ["Nope Group"], ws) == []
assert pm_sso.get_sso_roles(workspace_id=ws) == ["admin"]
# SSO-only switch is instance-wide (design §7.1).
_save_config(client, _saml_payload(cert_pem, sso_only=1))
assert pm_sso.is_sso_only_workspace(ws) is True
assert PermissionManager(uid).is_sso_only_workspace(ws) is True