"""Read-only SCIM 2.0 server — /scim/v2. Pins: * discovery: ServiceProviderConfig / ResourceTypes / Schemas behind a valid token (401 without), honest capability advertisement * Users list: RFC 7644 ListResponse shape, startIndex/count paging with correct totalResults, population filter applied * Users filters: userName eq / externalId eq / meta.lastModified gt; unsupported syntax → 400 invalidFilter (SCIM error schema) * Users/{id}: out-of-filter → 404 (no existence leak); ``active`` is False for a leaver visible through a widened filter * enterprise extension mapping (employeeNumber / department / costCenter / manager ref) * phoneNumbers emitted only with directory.pii:read * Groups: org nodes with in-filter members only; org:read required * writes: POST/PUT/PATCH/DELETE → 405 SCIM error with Allow: GET * scope split: persons:read alone cannot read Groups and vice versa """ from __future__ import annotations import uuid import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.identity.consumers.models import IdentityConsumer from app.identity.consumers.token_service import ConsumerTokenService from app.identity.person_models import ( IdentityOrgMembership, IdentityOrgNode, IdentityPerson, ) pytestmark = pytest.mark.asyncio BASE = "/scim/v2" def _h(token: str) -> dict: return {"Authorization": f"Bearer {token}"} async def _mk_consumer( db: AsyncSession, *, scopes: list[str], filter_config: dict | None = None, granted_fields: list[str] | None = None, ) -> str: owner = IdentityPerson( id=uuid.uuid4(), work_email=f"owner-{uuid.uuid4().hex[:8]}@pingcap.com", display_name="Owner", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:10]}", ) db.add(owner) await db.flush() consumer = IdentityConsumer( id=uuid.uuid4(), name="SCIM Puller", slug=f"scim-{uuid.uuid4().hex[:6]}", owner_person_id=owner.id, scopes=scopes, filter_config=filter_config or {}, granted_fields=granted_fields or [], ) db.add(consumer) await db.flush() await db.refresh(consumer) _row, plaintext = await ConsumerTokenService(db).mint( consumer=consumer, name="t", expires_in_days=None, created_by=None, ) await db.commit() return plaintext async def _mk_person( db: AsyncSession, *, email: str, status: str = "active", employee_no: str | None = None, dept: tuple[str, str] | None = None, mobile: str | None = None, manager_id: uuid.UUID | None = None, cost_center: str | None = None, ) -> uuid.UUID: pid = uuid.uuid4() db.add(IdentityPerson( id=pid, work_email=email, display_name=email.split("@")[0], first_name="First", last_name="Last", employment_status=status, source_system="successfactors", hris_person_id=f"sf-{email}", employee_no=employee_no, mobile_phone=mobile, manager_person_id=manager_id, cost_center=cost_center, )) await db.flush() if dept is not None: source_id, name = dept db.add(IdentityOrgMembership( id=uuid.uuid4(), person_id=pid, org_node_source_id=source_id, org_node_name=name, org_path=f"PingCAP / {name}", is_primary=True, )) await db.flush() return pid async def test_discovery_requires_token(client, db_session): r = await client.get(f"{BASE}/ServiceProviderConfig") assert r.status_code == 401 token = await _mk_consumer(db_session, scopes=["directory.persons:read"]) r = await client.get(f"{BASE}/ServiceProviderConfig", headers=_h(token)) assert r.status_code == 200 assert r.headers["content-type"].startswith("application/scim+json") cfg = r.json() assert cfg["patch"]["supported"] is False assert cfg["bulk"]["supported"] is False assert cfg["filter"]["supported"] is True r = await client.get(f"{BASE}/ResourceTypes", headers=_h(token)) ids = [x["id"] for x in r.json()["Resources"]] assert ids == ["User", "Group"] r = await client.get(f"{BASE}/Schemas", headers=_h(token)) assert r.status_code == 200 r = await client.get( f"{BASE}/ResourceTypes/Nope", headers=_h(token), ) assert r.status_code == 404 assert r.json()["schemas"] == ["urn:ietf:params:scim:api:messages:2.0:Error"] async def test_users_list_shape_and_paging(client, db_session): for i in range(5): await _mk_person( db_session, email=f"u{i}-{uuid.uuid4().hex[:5]}@pingcap.com", employee_no=f"P1000{i}", ) token = await _mk_consumer(db_session, scopes=["directory.persons:read"]) r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "startIndex": 1, "count": 2, }) assert r.status_code == 200 body = r.json() assert body["schemas"] == ["urn:ietf:params:scim:api:messages:2.0:ListResponse"] total = body["totalResults"] assert total >= 5 assert body["itemsPerPage"] == 2 assert body["startIndex"] == 1 first_ids = [u["id"] for u in body["Resources"]] r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "startIndex": 3, "count": 2, }) second_ids = [u["id"] for u in r.json()["Resources"]] assert not set(first_ids) & set(second_ids) # walk all pages → exactly totalResults unique users seen: list[str] = [] idx = 1 while True: r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "startIndex": idx, "count": 2, }) resources = r.json()["Resources"] if not resources: break seen.extend(u["id"] for u in resources) idx += len(resources) assert len(seen) == total assert len(set(seen)) == total async def test_users_filters(client, db_session): await _mk_person( db_session, email="findme@pingcap.com", employee_no="P31337", ) token = await _mk_consumer(db_session, scopes=["directory.persons:read"]) r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "filter": 'userName eq "findme@pingcap.com"', }) body = r.json() assert body["totalResults"] == 1 assert body["Resources"][0]["userName"] == "findme@pingcap.com" r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "filter": 'externalId eq "P31337"', }) assert r.json()["totalResults"] == 1 r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "filter": 'meta.lastModified gt "2099-01-01T00:00:00Z"', }) assert r.json()["totalResults"] == 0 # combined clause r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "filter": 'userName eq "findme@pingcap.com" and meta.lastModified gt "2000-01-01T00:00:00Z"', }) assert r.json()["totalResults"] == 1 # unsupported syntax → 400 invalidFilter for bad in ( 'userName co "find"', 'title eq "x"', "userName eq missing-quotes", 'a eq "1" and b eq "2" and c eq "3"', ): r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "filter": bad, }) assert r.status_code == 400, bad assert r.json()["scimType"] == "invalidFilter" async def test_user_detail_filter_404_active_and_enterprise(client, db_session): mgr = await _mk_person( db_session, email="mgr@pingcap.com", employee_no="P00001", dept=("D100", "Engineering"), ) pid = await _mk_person( db_session, email="dev@pingcap.com", employee_no="P00002", dept=("D100", "Engineering"), manager_id=mgr, cost_center="CC42", ) outsider = await _mk_person( db_session, email="out@pingcap.com", dept=("D200", "Sales"), ) leaver = await _mk_person( db_session, email="gone@pingcap.com", status="left", dept=("D100", "Engineering"), ) token = await _mk_consumer( db_session, scopes=["directory.persons:read"], filter_config={ "departments": ["D100"], "include_subtree": False, "employment_statuses": ["active", "left"], }, ) r = await client.get(f"{BASE}/Users/{pid}", headers=_h(token)) assert r.status_code == 200 u = r.json() ent = u["urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"] assert ent["employeeNumber"] == "P00002" assert ent["department"] == "Engineering" assert ent["costCenter"] == "CC42" assert ent["manager"]["value"] == str(mgr) assert u["active"] is True assert u["meta"]["resourceType"] == "User" r = await client.get(f"{BASE}/Users/{leaver}", headers=_h(token)) assert r.status_code == 200 assert r.json()["active"] is False r = await client.get(f"{BASE}/Users/{outsider}", headers=_h(token)) assert r.status_code == 404 r = await client.get(f"{BASE}/Users/not-a-uuid", headers=_h(token)) assert r.status_code == 404 async def test_phone_numbers_pii_gated(client, db_session): pid = await _mk_person( db_session, email="phone@pingcap.com", mobile="+8613812345678", ) plain = await _mk_consumer(db_session, scopes=["directory.persons:read"]) r = await client.get(f"{BASE}/Users/{pid}", headers=_h(plain)) assert "phoneNumbers" not in r.json() pii = await _mk_consumer( db_session, scopes=["directory.persons:read", "directory.pii:read"], ) r = await client.get(f"{BASE}/Users/{pid}", headers=_h(pii)) assert r.json()["phoneNumbers"] == [ {"value": "+8613812345678", "type": "mobile"}, ] async def test_groups_members_respect_filter_and_scope(client, db_session): node_id = uuid.uuid4() db_session.add(IdentityOrgNode( id=node_id, org_node_source_id="D100", org_node_name="Engineering", org_path="PingCAP / Engineering", )) await db_session.flush() await _mk_person( db_session, email="g-eng@pingcap.com", dept=("D100", "Engineering"), employee_no="P20001", ) await _mk_person( db_session, email="g-eng-no@pingcap.com", dept=("D100", "Engineering"), employee_no="X-SERVICE", ) await db_session.commit() # persons scope alone cannot read Groups persons_only = await _mk_consumer( db_session, scopes=["directory.persons:read"], ) r = await client.get(f"{BASE}/Groups", headers=_h(persons_only)) assert r.status_code == 403 token = await _mk_consumer( db_session, scopes=["directory.persons:read", "directory.org:read"], # employee_no gate excludes the service account from members filter_config={"employee_no_pattern": r"^P\d+$"}, ) r = await client.get(f"{BASE}/Groups", headers=_h(token), params={ "filter": 'displayName eq "Engineering"', }) assert r.status_code == 200 body = r.json() assert body["totalResults"] == 1 grp = body["Resources"][0] assert grp["externalId"] == "D100" member_displays = [m["display"] for m in grp["members"]] assert "g-eng" in member_displays assert "g-eng-no" not in member_displays r = await client.get(f"{BASE}/Groups/{node_id}", headers=_h(token)) assert r.status_code == 200 assert r.json()["displayName"] == "Engineering" async def test_filter_value_containing_and_parses(client, db_session): """A quoted value containing ' and ' must parse (quote-aware filter, review fix 2026-06-11).""" node_id = uuid.uuid4() db_session.add(IdentityOrgNode( id=node_id, org_node_source_id="D300", org_node_name="Research and Development", org_path="PingCAP / Research and Development", )) await db_session.commit() token = await _mk_consumer(db_session, scopes=["directory.org:read"]) r = await client.get(f"{BASE}/Groups", headers=_h(token), params={ "filter": 'displayName eq "Research and Development"', }) assert r.status_code == 200, r.text assert r.json()["totalResults"] == 1 assert r.json()["Resources"][0]["displayName"] == "Research and Development" async def test_get_unknown_path_is_scim_404(client, db_session): """GET on an unknown SCIM path → 404 SCIM error (not a reversed-Allow 405), and it requires auth like everything else.""" r = await client.get(f"{BASE}/Bulk") assert r.status_code == 401 # auth first, even on unknown paths token = await _mk_consumer(db_session, scopes=["directory.persons:read"]) r = await client.get(f"{BASE}/Bulk", headers=_h(token)) assert r.status_code == 404 assert r.json()["schemas"] == ["urn:ietf:params:scim:api:messages:2.0:Error"] async def test_out_of_filter_manager_omitted(client, db_session): """A manager outside the consumer's filter must not leak through the SCIM enterprise manager block (displayName = full name).""" mgr = await _mk_person( db_session, email="scim-boss@pingcap.com", employee_no="P00200", dept=("D200", "Sales"), ) pid = await _mk_person( db_session, email="scim-report@pingcap.com", employee_no="P00201", dept=("D100", "Engineering"), manager_id=mgr, ) token = await _mk_consumer( db_session, scopes=["directory.persons:read"], filter_config={"departments": ["D100"], "include_subtree": False}, ) r = await client.get(f"{BASE}/Users/{pid}", headers=_h(token)) assert r.status_code == 200 ent = r.json()["urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"] assert "manager" not in ent async def test_writes_rejected_405(client, db_session): token = await _mk_consumer( db_session, scopes=["directory.persons:read", "directory.org:read"], ) for method, path in ( ("post", "/Users"), ("put", "/Users/00000000-0000-0000-0000-000000000000"), ("patch", "/Users/00000000-0000-0000-0000-000000000000"), ("delete", "/Groups/00000000-0000-0000-0000-000000000000"), ("post", "/Bulk"), ): r = await getattr(client, method)( f"{BASE}{path}", headers=_h(token), **({"json": {}} if method in ("post", "put", "patch") else {}), ) assert r.status_code == 405, (method, path) assert r.headers.get("allow") == "GET" assert r.json()["schemas"] == [ "urn:ietf:params:scim:api:messages:2.0:Error", ] # writes are auth-gated too r = await client.post(f"{BASE}/Users", json={}) assert r.status_code == 401 async def test_scim_out_of_filter_uuid_is_404_like_missing(client, db_session): """An out-of-filter real user and a non-existent UUID both 404 with the identical SCIM Error body — no existence oracle (the timing parity fix is behavioural-tested here; we assert response equality, the equal-work scope resolution backs the timing side).""" import uuid as _uuid outsider = await _mk_person( db_session, email="scim-out@pingcap.com", dept=("D200", "Sales"), ) token = await _mk_consumer( db_session, scopes=["directory.persons:read"], filter_config={"departments": ["D100"], "include_subtree": False}, ) r_out = await client.get(f"{BASE}/Users/{outsider}", headers=_h(token)) r_missing = await client.get( f"{BASE}/Users/{_uuid.uuid4()}", headers=_h(token), ) assert r_out.status_code == 404 assert r_missing.status_code == 404 assert r_out.json()["schemas"] == r_missing.json()["schemas"] assert r_out.json()["status"] == r_missing.json()["status"] == "404" async def test_scim_location_uses_configured_origin_not_host_header( client, db_session, ): """meta.location must derive from the configured public origin, not a client-supplied Host header (security review 2026-06-11).""" from app.core.config import settings pid = await _mk_person(db_session, email="loc@pingcap.com") token = await _mk_consumer(db_session, scopes=["directory.persons:read"]) original = settings.CORS_ORIGINS settings.CORS_ORIGINS = ["https://helix.example.net"] try: r = await client.get( f"{BASE}/Users/{pid}", headers={**_h(token), "Host": "evil.attacker.test"}, ) assert r.status_code == 200 loc = r.json()["meta"]["location"] assert loc.startswith("https://helix.example.net/scim/v2/Users/") assert "evil.attacker.test" not in loc finally: settings.CORS_ORIGINS = original async def test_scim_users_respect_feed_safety_horizon(client, db_session): """SCIM Users honour the same visibility safety lag as the REST keyset feed — a just-written user is held back so a lastModified incremental can't step over it (logic review round 2, 2026-06-11).""" from sqlalchemy import text as _text from app.core.config import settings pid_old = await _mk_person(db_session, email="scim-old@pingcap.com") await db_session.execute( _text("UPDATE identity_persons SET updated_at = :u WHERE id = :i"), {"u": "2026-06-01 00:00:00", "i": str(pid_old)}, ) await _mk_person(db_session, email="scim-fresh@pingcap.com") # now() await db_session.commit() token = await _mk_consumer(db_session, scopes=["directory.persons:read"]) settings.DIRECTORY_API_FEED_SAFETY_LAG_SECONDS = 30 try: r = await client.get(f"{BASE}/Users", headers=_h(token), params={ "count": 100, }) usernames = [u["userName"] for u in r.json()["Resources"]] assert "scim-old@pingcap.com" in usernames assert "scim-fresh@pingcap.com" not in usernames finally: settings.DIRECTORY_API_FEED_SAFETY_LAG_SECONDS = 0 async def test_scim_users_respect_field_level_grant(client, db_session): """2026-07-07 SECURITY: a consumer narrowed BELOW its persons bundle (display_name + work_email only) must NOT receive the withheld internal-tier fields via SCIM — the least-privilege field grant is protocol-agnostic and must match the REST projection. Previously SCIM emitted every attribute unconditionally, so a narrowed consumer could read employee_no / cost_center / department / manager / location by switching from /directory/persons to /scim/v2/Users.""" pid = await _mk_person( db_session, email="narrow@pingcap.com", employee_no="E123", cost_center="CC-9", dept=("d1", "Platform"), ) token = await _mk_consumer( db_session, scopes=["directory.persons:read"], granted_fields=["display_name", "work_email"], ) r = await client.get(f"{BASE}/Users/{pid}", headers=_h(token)) assert r.status_code == 200, r.text body = r.json() # Granted core fields present. assert body.get("displayName") == "narrow" assert body.get("userName") == "narrow@pingcap.com" # Withheld fields absent (these leaked before the fix). assert "externalId" not in body # employee_no not granted assert "title" not in body # job_title not granted assert "addresses" not in body # work_location not granted ENT = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User" assert ENT not in body or not body[ENT], ( "enterprise (employeeNumber/costCenter/department/manager) must be " "withheld when not granted" )