"""End-to-end unit tests: real bridge HTTP server against a fake NexusCRM. The fake implements the exact public-API subset the bridge uses (oauth token, SOQL query, sobject create/patch) with an in-memory User table, so every SCIM flow is exercised over real HTTP without a live platform. """ import json import re import socket import threading import unittest import urllib.parse import urllib.request import urllib.error from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import sys import tempfile from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import scim_bridge # noqa: E402 CLIENT_ID, CLIENT_SECRET, TOKEN = "helix-scim", "s3cret", "tok-1" class FakeNexus(BaseHTTPRequestHandler): users: dict = {} next_id = [1] token_calls = [0] token_success = [0] fail_patch = [False, 400] profile_grant_plane = [None] permission_queries = [] def log_message(self, *a): # quiet pass def _json(self, status, payload): raw = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def _authed(self): return self.headers.get("Authorization") == f"Bearer {TOKEN}" def do_POST(self): length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) if self.path == "/services/oauth2/token": form = urllib.parse.parse_qs(body.decode()) self.token_calls[0] += 1 if (form.get("grant_type") == ["client_credentials"] and form.get("client_id") == [CLIENT_ID] and form.get("client_secret") == [CLIENT_SECRET]): self.token_success[0] += 1 return self._json(200, {"access_token": TOKEN, "expires_in": 7200}) return self._json(401, {"error": "invalid_client"}) if not self._authed(): return self._json(401, {"error": "unauthorized"}) if self.path.endswith("/sobjects/User"): fields = json.loads(body) known = {"Username", "Email", "FirstName", "LastName", "Alias", "ProfileId", "TimeZoneSidKey", "LocaleSidKey", "EmailEncodingKey", "LanguageLocaleKey", "IsActive", "Title", "Phone", "MobilePhone", "Department", "Division", "Company", "ManagerId"} unknown = set(fields) - known if unknown: return self._json(400, [{"errorCode": "INVALID_FIELD", "message": sorted(unknown)}]) for required in ("Username", "Email", "LastName", "Alias", "ProfileId", "TimeZoneSidKey", "LocaleSidKey", "EmailEncodingKey", "LanguageLocaleKey"): if not fields.get(required): return self._json(400, [{"errorCode": "REQUIRED_FIELD_MISSING", "message": required}]) if "Password" in fields: return self._json(400, [{"errorCode": "TEST_PASSWORD_SENT"}]) if any(u["Username"] == fields["Username"] for u in self.users.values()): return self._json(400, [{"errorCode": "DUPLICATE_USERNAME"}]) uid = f"005U{self.next_id[0]:06d}" self.next_id[0] += 1 record = {"Id": uid, "IsActive": True, "FirstName": "", "LastName": ""} record.update(fields) self.users[uid] = record return self._json(201, {"id": uid, "success": True}) return self._json(404, {"error": self.path}) def do_PATCH(self): if not self._authed(): return self._json(401, {}) if self.fail_patch[0]: return self._json(self.fail_patch[1], {"error": "injected"}) match = re.search(r"/sobjects/User/([^/]+)$", self.path) if not match or match.group(1) not in self.users: return self._json(404, {}) length = int(self.headers.get("Content-Length") or 0) self.users[match.group(1)].update(json.loads(self.rfile.read(length))) self.send_response(204) self.send_header("Content-Length", "0") self.end_headers() def do_GET(self): if not self._authed(): return self._json(401, {}) parsed = urllib.parse.urlparse(self.path) if "REDIRECTME" in parsed.query: self.send_response(302) self.send_header("Location", "http://evil.example/steal") self.send_header("Content-Length", "0") self.end_headers() return if "/query" not in parsed.path: return self._json(404, {}) soql = urllib.parse.parse_qs(parsed.query)["q"][0] records = [] m = re.search(r"FROM Profile WHERE Name = '([^']+)'", soql) if m: records = ( [{ "Id": "00ePROFILE01", "IsActive": True, "IsSystem": False, "UserType": "Standard", }] if m.group(1) == scim_bridge.SCIM_ZERO_ACCESS_PROFILE else [] ) if "FROM PermissionSet " in soql: records = [{"Id": "0PSPROFILE01"}] grant_query_contracts = { "_System_ProfilePermission": ("ProfileId =", "IsEnabled = true"), "_System_PermissionSetPermission": ("PermissionSetId =", "IsEnabled = true"), "ObjectPermissions": ( "ParentId =", "IsDeleted = false", "PermissionsRead = true", "PermissionsCreate = true", "PermissionsEdit = true", "PermissionsDelete = true", "PermissionsViewAllRecords = true", "PermissionsModifyAllRecords = true", ), "FieldPermissions": ( "ParentId =", "IsDeleted = false", "PermissionsRead = true", "PermissionsEdit = true", ), "PermissionSetCustomPermission": ("PermissionSetId =",), "ProfileCustomPermission": ("ProfileId =",), "PermissionSetRecordTypeVisibility": ("PermissionSetId =", "IsVisible = true"), } for grant_plane, required_fragments in grant_query_contracts.items(): if f"FROM {grant_plane} " in soql: self.permission_queries.append(soql) if not all(fragment in soql for fragment in required_fragments): return self._json(400, [{"errorCode": "INVALID_FIELD"}]) records = ( [{"Id": "0GRANT01"}] if self.profile_grant_plane[0] == grant_plane else [] ) m = re.search(r"FROM User WHERE Username = '([^']+)'", soql) if m: records = [dict(u) for u in self.users.values() if u["Username"] == m.group(1)] m = re.search(r"FROM User WHERE Email = '([^']+)' OR Username = '([^']+)'", soql) if m: records = [dict(u) for u in self.users.values() if u.get("Email") == m.group(1) or u["Username"] == m.group(2)] m = re.search(r"FROM User WHERE Id = '([^']+)'", soql) if m and m.group(1) in self.users: records = [dict(self.users[m.group(1)])] return self._json(200, {"totalSize": len(records), "records": records}) def scim(method, path, body=None, token=f"{CLIENT_ID}.{CLIENT_SECRET}"): url = f"http://127.0.0.1:{scim._port}{path}" data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", f"Bearer {token}") if data: req.add_header("Content-Type", "application/scim+json") try: with urllib.request.urlopen(req, timeout=10) as resp: raw = resp.read() return resp.status, (json.loads(raw) if raw else None) except urllib.error.HTTPError as err: raw = err.read() return err.code, (json.loads(raw) if raw else None) class BridgeE2ETest(unittest.TestCase): @classmethod def setUpClass(cls): cls.nexus = ThreadingHTTPServer(("127.0.0.1", 0), FakeNexus) threading.Thread(target=cls.nexus.serve_forever, daemon=True).start() config = scim_bridge.BridgeConfig({ "BRIDGE_NEXUS_BASE_URL": f"http://127.0.0.1:{cls.nexus.server_port}", "BRIDGE_LISTEN_PORT": "0", "BRIDGE_DEACTIVATION_MAX": "2", "BRIDGE_EXEMPT_EMAILS": "ceo@pingcap.com,admin@example.invalid", "BRIDGE_BOOTSTRAP_ADMIN": "admin@example.invalid", "BRIDGE_STATE_FILE": tempfile.mktemp(prefix="scim-state-")}) cls.bridge = scim_bridge.serve(config) threading.Thread(target=cls.bridge.serve_forever, daemon=True).start() scim._port = cls.bridge.server_port @classmethod def tearDownClass(cls): cls.bridge.shutdown() cls.nexus.shutdown() def payload(self, email, **kw): body = {"schemas": [scim_bridge.SCIM_USER_SCHEMA], "userName": email, "active": True, "name": {"givenName": "Wei", "familyName": "Zhang"}, "emails": [{"value": email, "primary": True}]} body.update(kw) return body def test_01_bad_credential_rejected(self): status, body = scim("GET", "/scim/v2/Users?filter=" + urllib.parse.quote( 'userName eq "x@pingcap.com"'), token="helix-scim.WRONG") self.assertEqual(status, 401) self.assertIn(scim_bridge.SCIM_ERROR_SCHEMA, body["schemas"]) def test_02_create_sets_defaults_never_password(self): status, body = scim("POST", "/scim/v2/Users", self.payload( "wei.zhang@pingcap.com", title="Engineer", phoneNumbers=[{"type": "mobile", "value": "+86-1"}], **{scim_bridge.SCIM_ENTERPRISE_SCHEMA: { "department": "PingCAP / Engineering", "organization": "PingCAP"}})) self.assertEqual(status, 201, body) self.assertTrue(body["id"].startswith("005U")) record = FakeNexus.users[body["id"]] self.assertEqual(record["ProfileId"], "00ePROFILE01") self.assertEqual(record["Department"], "PingCAP / Engineering") self.assertEqual(record["Company"], "PingCAP") self.assertEqual(record["MobilePhone"], "+86-1") self.assertNotIn("Password", record) # SSO-only def test_03_create_is_idempotent_converges(self): status, body = scim("POST", "/scim/v2/Users", self.payload("wei.zhang@pingcap.com", title="Staff Engineer")) self.assertEqual(status, 200, body) # converged, not duplicated self.assertEqual( sum(1 for u in FakeNexus.users.values() if u["Username"] == "wei.zhang@pingcap.com"), 1) def test_04_filter_find_then_put_update_and_manager(self): scim("POST", "/scim/v2/Users", self.payload("mgr@pingcap.com", name={"givenName": "Yu", "familyName": "Dong"})) status, body = scim("GET", "/scim/v2/Users?filter=" + urllib.parse.quote( 'userName eq "wei.zhang@pingcap.com"')) self.assertEqual(status, 200) self.assertEqual(body["totalResults"], 1) uid = body["Resources"][0]["id"] status, body = scim("PUT", f"/scim/v2/Users/{uid}", self.payload( "wei.zhang@pingcap.com", **{scim_bridge.SCIM_ENTERPRISE_SCHEMA: {"manager": {"value": "mgr@pingcap.com"}}})) self.assertEqual(status, 200, body) manager_id = FakeNexus.users[uid]["ManagerId"] self.assertEqual(FakeNexus.users[manager_id]["Username"], "mgr@pingcap.com") def test_05_patch_deactivate_and_delete_maps_to_deactivate(self): status, body = scim("POST", "/scim/v2/Users", self.payload("leaver@pingcap.com")) uid = body["id"] status, body = scim("PATCH", f"/scim/v2/Users/{uid}", { "schemas": [scim_bridge.SCIM_PATCHOP_SCHEMA], "Operations": [{"op": "replace", "path": "active", "value": False}]}) self.assertEqual(status, 200, body) self.assertFalse(FakeNexus.users[uid]["IsActive"]) status, body = scim("POST", "/scim/v2/Users", self.payload("leaver2@pingcap.com")) uid2 = body["id"] status, _ = scim("DELETE", f"/scim/v2/Users/{uid2}") self.assertEqual(status, 204) self.assertFalse(FakeNexus.users[uid2]["IsActive"]) # deactivated, not deleted self.assertIn(uid2, FakeNexus.users) def test_06_exemption_and_breaker(self): status, body = scim("POST", "/scim/v2/Users", self.payload("ceo@pingcap.com")) status, body = scim("DELETE", f"/scim/v2/Users/{body['id']}") self.assertEqual(status, 403) # exempt list blocks deactivation status, body = scim("POST", "/scim/v2/Users", self.payload("b3@pingcap.com")) status, body = scim("DELETE", f"/scim/v2/Users/{body['id']}") self.assertEqual(status, 429) # breaker: max=2 already consumed in test_05 def test_06b_wrong_secret_cannot_ride_cached_token(self): # A valid request has already cached a token for this client_id. # The same client_id with a WRONG secret must be re-authenticated # against NexusCRM and rejected, never served from the cache. status, body = scim("GET", "/scim/v2/Users?filter=" + urllib.parse.quote( 'userName eq "wei.zhang@pingcap.com"'), token=f"{CLIENT_ID}.WRONG-SECRET") self.assertEqual(status, 401, body) def test_06c_unknown_manager_is_rejected_not_cleared(self): status, body = scim("GET", "/scim/v2/Users?filter=" + urllib.parse.quote( 'userName eq "wei.zhang@pingcap.com"')) uid = body["Resources"][0]["id"] before = FakeNexus.users[uid].get("ManagerId") status, body = scim("PUT", f"/scim/v2/Users/{uid}", self.payload( "wei.zhang@pingcap.com", **{scim_bridge.SCIM_ENTERPRISE_SCHEMA: { "manager": {"value": "ghost@pingcap.com"}}})) self.assertEqual(status, 400, body) # rejected, not silently cleared self.assertEqual(FakeNexus.users[uid].get("ManagerId"), before) def test_06d_plaintext_nonloopback_upstream_refused(self): with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_NEXUS_BASE_URL": "http://crm.example.com"}) scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", # loopback ok for dev "BRIDGE_NEXUS_BASE_URL": "http://127.0.0.1:3001"}) def test_06e_exemption_matches_email_when_username_differs(self): status, body = scim("POST", "/scim/v2/Users", self.payload("cfo.login@pingcap.com")) uid = body["id"] FakeNexus.users[uid]["Email"] = "ceo@pingcap.com" # exempt by Email only status, body = scim("DELETE", f"/scim/v2/Users/{uid}") self.assertEqual(status, 403, body) self.assertTrue(FakeNexus.users[uid]["IsActive"]) def test_06f_malformed_active_is_400_not_deactivate(self): status, body = scim("GET", "/scim/v2/Users?filter=" + urllib.parse.quote( 'userName eq "wei.zhang@pingcap.com"')) uid = body["Resources"][0]["id"] status, body = scim("PATCH", f"/scim/v2/Users/{uid}", { "schemas": [scim_bridge.SCIM_PATCHOP_SCHEMA], "Operations": [{"op": "replace", "path": "active", "value": "garbage"}]}) self.assertEqual(status, 400, body) self.assertTrue(FakeNexus.users[uid]["IsActive"]) # untouched def test_06g_breaker_config_fails_closed(self): for bad in ("0", "-5"): with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_DEACTIVATION_WINDOW_SECONDS": bad}) with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_DEACTIVATION_MAX": bad}) def test_06h_uppercase_plaintext_scheme_refused(self): with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_NEXUS_BASE_URL": "HTTP://crm.example.com"}) def test_06i_exempt_list_is_required(self): with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_NEXUS_BASE_URL": "http://127.0.0.1:1"}) def test_06i2_nonzero_onboarding_profile_is_rejected(self): with self.assertRaisesRegex(ValueError, "must be SCIM Zero Access"): scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_DEFAULT_PROFILE_NAME": "Standard User", }) def test_06i3_permission_plane_drift_blocks_first_time_create(self): FakeNexus.permission_queries.clear() grant_planes = ( "_System_ProfilePermission", "_System_PermissionSetPermission", "ObjectPermissions", "FieldPermissions", "PermissionSetCustomPermission", "ProfileCustomPermission", "PermissionSetRecordTypeVisibility", ) for index, grant_plane in enumerate(grant_planes): with self.subTest(grant_plane=grant_plane): FakeNexus.profile_grant_plane[0] = grant_plane email = f"drift-{index}@pingcap.com" before = len(FakeNexus.users) try: status, body = scim( "POST", "/scim/v2/Users", self.payload(email), ) finally: FakeNexus.profile_grant_plane[0] = None self.assertEqual(status, 503, body) self.assertEqual(len(FakeNexus.users), before) self.assertTrue(any( "FROM _System_ProfilePermission " in query and "IsEnabled = true" in query and " Enabled = true" not in query for query in FakeNexus.permission_queries )) plane_contracts = { "ObjectPermissions": ( "IsDeleted = false", "PermissionsRead = true", "PermissionsCreate = true", "PermissionsEdit = true", "PermissionsDelete = true", "PermissionsViewAllRecords = true", "PermissionsModifyAllRecords = true", ), "FieldPermissions": ( "IsDeleted = false", "PermissionsRead = true", "PermissionsEdit = true", ), } for plane, required_fragments in plane_contracts.items(): self.assertTrue(any( f"FROM {plane} " in query and "ParentId =" in query and "PermissionSetId =" not in query and all(fragment in query for fragment in required_fragments) for query in FakeNexus.permission_queries )) def test_06i4_profile_shape_and_owner_drift_fail_closed(self): config = scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_STATE_FILE": tempfile.mktemp(prefix="scim-state-"), }) service = scim_bridge.ScimBridgeService(config) valid = { "Id": "00ePROFILE01", "IsActive": True, "IsSystem": False, "UserType": "Standard", } bad_profile_sets = ( [], [valid, dict(valid)], [{**valid, "IsActive": False}], [{**valid, "IsSystem": True}], [{**valid, "UserType": "PowerPartner"}], ) for profiles in bad_profile_sets: with self.subTest(profiles=profiles): client = type( "ProfileClient", (), {"query": lambda _self, _query, rows=profiles: rows}, )() with self.assertRaises(scim_bridge.BridgeError) as caught: service._profile_id(client) self.assertEqual(caught.exception.status, 503) for permission_sets in ([], [{"Id": "a"}, {"Id": "b"}]): with self.subTest(permission_sets=permission_sets): responses = iter(([valid], permission_sets)) client = type( "ProfileOwnerClient", (), {"query": lambda _self, _query: next(responses)}, )() with self.assertRaises(scim_bridge.BridgeError) as caught: service._profile_id(client) self.assertEqual(caught.exception.status, 503) def test_06j_put_clears_absent_managed_attributes(self): status, body = scim("POST", "/scim/v2/Users", self.payload("clearer@pingcap.com", title="Director")) uid = body["id"] self.assertEqual(FakeNexus.users[uid]["Title"], "Director") status, body = scim("PUT", f"/scim/v2/Users/{uid}", self.payload("clearer@pingcap.com")) # no title now self.assertEqual(status, 200, body) self.assertIsNone(FakeNexus.users[uid]["Title"]) # truly replaced def test_06k_create_without_username_uses_primary_email(self): status, body = scim("POST", "/scim/v2/Users", { "schemas": [scim_bridge.SCIM_USER_SCHEMA], "active": True, "name": {"familyName": "NoUserName"}, "emails": [{"value": "only.email@pingcap.com", "primary": True}]}) self.assertEqual(status, 201, body) self.assertEqual(FakeNexus.users[body["id"]]["Username"], "only.email@pingcap.com") def test_06l_malformed_active_on_post_is_400(self): status, body = scim("POST", "/scim/v2/Users", self.payload("wei.zhang@pingcap.com", active="garbage")) self.assertEqual(status, 400, body) def test_06m_nonloopback_listener_requires_tls_ack(self): with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_LISTEN_HOST": "0.0.0.0"}) scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_LISTEN_HOST": "0.0.0.0", "BRIDGE_TLS_TERMINATED": "1"}) def test_06n_readback_includes_managed_attributes(self): status, body = scim("GET", "/scim/v2/Users?filter=" + urllib.parse.quote( 'userName eq "wei.zhang@pingcap.com"')) uid = body["Resources"][0]["id"] FakeNexus.users[uid].update({"Title": "Engineer", "Department": "Eng", "Company": "PingCAP", "MobilePhone": "+86-1"}) status, body = scim("GET", f"/scim/v2/Users/{uid}") self.assertEqual(body.get("title"), "Engineer") ent = body.get(scim_bridge.SCIM_ENTERPRISE_SCHEMA) or {} self.assertEqual(ent.get("department"), "Eng") self.assertEqual(ent.get("organization"), "PingCAP") self.assertIn({"type": "mobile", "value": "+86-1"}, body.get("phoneNumbers") or []) def test_06o_post_never_replayed_on_401(self): # Unit-level: a POST issued with an expired token is NOT retried with a # fresh one (a replay could duplicate the user if NexusCRM committed # before the 401 surfaced); idempotent GETs do retry and heal the cache. config = scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_NEXUS_BASE_URL": f"http://127.0.0.1:{self.nexus.server_port}"}) client = scim_bridge.NexusClient(config, CLIENT_ID, CLIENT_SECRET) scim_bridge.NexusClient._token_cache[client._cache_key] = ("bad-token", 10**12) before = len(FakeNexus.users) status, _ = client.call("POST", "/sobjects/User", {"Username": "x@y.z"}) self.assertEqual(status, 401) # surfaced, not replayed self.assertEqual(len(FakeNexus.users), before) # nothing created status, _ = client.call("GET", "/query?q=" + urllib.parse.quote( "SELECT Id FROM User WHERE Username = 'wei.zhang@pingcap.com' LIMIT 1")) self.assertEqual(status, 200) # cache healed; idempotent retry works def test_06p_exempt_user_cannot_be_renamed(self): uid = next(u for u, r in FakeNexus.users.items() if r.get("Email") == "ceo@pingcap.com") status, body = scim("PUT", f"/scim/v2/Users/{uid}", self.payload("sneaky.rename@pingcap.com")) self.assertEqual(status, 403, body) def test_06q_externalid_filter_rejected(self): status, body = scim("GET", "/scim/v2/Users?filter=" + urllib.parse.quote( 'externalId eq "x@pingcap.com"')) self.assertEqual(status, 400, body) def test_06r_put_replaces_names_never_stale(self): status, body = scim("POST", "/scim/v2/Users", self.payload("renamer@pingcap.com")) uid = body["id"] status, body = scim("PUT", f"/scim/v2/Users/{uid}", { "schemas": [scim_bridge.SCIM_USER_SCHEMA], "userName": "renamer@pingcap.com", "active": True}) # no name at all self.assertEqual(status, 200, body) self.assertIsNone(FakeNexus.users[uid]["FirstName"]) # cleared self.assertEqual(FakeNexus.users[uid]["LastName"], "renamer") # re-derived def test_06s_patch_path_specific_ops(self): status, body = scim("POST", "/scim/v2/Users", self.payload("pather@pingcap.com")) uid = body["id"] status, body = scim("PATCH", f"/scim/v2/Users/{uid}", { "schemas": [scim_bridge.SCIM_PATCHOP_SCHEMA], "Operations": [ {"op": "replace", "path": "title", "value": "VP"}, {"op": "replace", "path": "name.givenName", "value": "Ming"}, {"op": "replace", "path": scim_bridge.SCIM_ENTERPRISE_SCHEMA + ":department", "value": "Sales"}, {"op": "replace", "path": 'phoneNumbers[type eq "mobile"].value', "value": "+86-9"}]}) self.assertEqual(status, 200, body) record = FakeNexus.users[uid] self.assertEqual((record["Title"], record["FirstName"], record["Department"], record["MobilePhone"]), ("VP", "Ming", "Sales", "+86-9")) status, body = scim("PATCH", f"/scim/v2/Users/{uid}", { "schemas": [scim_bridge.SCIM_PATCHOP_SCHEMA], "Operations": [{"op": "remove", "path": "title"}]}) self.assertEqual(status, 200, body) self.assertIsNone(FakeNexus.users[uid]["Title"]) def test_06t_breaker_budget_survives_restart(self): state = tempfile.mktemp(prefix="scim-state-") config = scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_DEACTIVATION_MAX": "2", "BRIDGE_STATE_FILE": state}) service = scim_bridge.ScimBridgeService(config) service._guard_deactivation({"Username": "u1@x.co"}) service._guard_deactivation({"Username": "u2@x.co"}) reborn = scim_bridge.ScimBridgeService(config) # simulated restart with self.assertRaises(scim_bridge.BridgeError) as caught: reborn._guard_deactivation({"Username": "u3@x.co"}) self.assertEqual(caught.exception.status, 429) # budget persisted def test_06u_mutations_never_retried_on_401(self): config = scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_STATE_FILE": tempfile.mktemp(prefix="scim-state-"), "BRIDGE_NEXUS_BASE_URL": f"http://127.0.0.1:{self.nexus.server_port}"}) client = scim_bridge.NexusClient(config, CLIENT_ID, CLIENT_SECRET) uid = next(iter(FakeNexus.users)) before = dict(FakeNexus.users[uid]) scim_bridge.NexusClient._token_cache[client._cache_key] = ("bad-token", 10**12) status, _ = client.call("PATCH", f"/sobjects/User/{uid}", {"Title": "X"}) self.assertEqual(status, 401) # surfaced, not replayed self.assertEqual(FakeNexus.users[uid], before) # untouched def _direct_client(self, **extra): env = {"BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_STATE_FILE": tempfile.mktemp(prefix="scim-state-"), "BRIDGE_NEXUS_BASE_URL": f"http://127.0.0.1:{self.nexus.server_port}"} env.update(extra) config = scim_bridge.BridgeConfig(env) return config, scim_bridge.NexusClient(config, CLIENT_ID, CLIENT_SECRET) def test_06v_redirects_never_followed(self): _, client = self._direct_client() status, _ = client.call("GET", "/query?q=REDIRECTME") self.assertEqual(status, 302) # surfaced as error, bearer not forwarded def test_06w_corrupt_state_file_fails_closed(self): state = tempfile.mktemp(prefix="scim-state-") with open(state, "w") as handle: handle.write("{not json") config = scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "a@b.c", "BRIDGE_BOOTSTRAP_ADMIN": "a@b.c", "BRIDGE_STATE_FILE": state}) with self.assertRaises(ValueError): scim_bridge.ScimBridgeService(config) def test_06x_oversized_and_empty_bodies_rejected(self): with socket.create_connection(("127.0.0.1", scim._port), timeout=10) as sock: sock.sendall( b"POST /scim/v2/Users HTTP/1.1\r\nHost: x\r\n" b"Authorization: Bearer " + f"{CLIENT_ID}.{CLIENT_SECRET}".encode() + b"\r\nContent-Length: 5000000\r\n\r\n") reply = sock.recv(4096).decode("utf-8", "replace") self.assertIn(" 413 ", reply.splitlines()[0]) req = urllib.request.Request( f"http://127.0.0.1:{scim._port}/scim/v2/Users/005U000001", data=b"", method="PUT") req.add_header("Authorization", f"Bearer {CLIENT_ID}.{CLIENT_SECRET}") try: with urllib.request.urlopen(req, timeout=10) as resp: status = resp.status except urllib.error.HTTPError as err: status = err.code self.assertEqual(status, 400) # empty PUT must not wipe fields def test_06y_recreate_does_not_resurrect_deactivated_user(self): status, body = scim("POST", "/scim/v2/Users", self.payload("ghosted@pingcap.com")) uid = body["id"] FakeNexus.users[uid]["IsActive"] = False # offboarded out-of-band status, body = scim("POST", "/scim/v2/Users", { "schemas": [scim_bridge.SCIM_USER_SCHEMA], "userName": "ghosted@pingcap.com"}) # no active attribute self.assertEqual(status, 200, body) self.assertFalse(FakeNexus.users[uid]["IsActive"]) # stays deprovisioned def test_06z_failed_deactivation_does_not_burn_budget(self): config, client = self._direct_client(BRIDGE_DEACTIVATION_MAX="1") service = scim_bridge.ScimBridgeService(config) status, body = scim("POST", "/scim/v2/Users", self.payload("flaky@pingcap.com")) uid = body["id"] FakeNexus.fail_patch[0], FakeNexus.fail_patch[1] = True, 400 try: with self.assertRaises(scim_bridge.BridgeError) as caught: service.delete_user(client, uid) self.assertEqual(caught.exception.status, 502) finally: FakeNexus.fail_patch[0] = False service.delete_user(client, uid) # 4xx rejection returned the budget self.assertFalse(FakeNexus.users[uid]["IsActive"]) def test_06z2_ambiguous_5xx_failure_keeps_budget(self): # A 5xx may have committed the deactivation before erroring, so the # reservation must be kept -- committed deactivations can never slip # past the breaker budget. config, client = self._direct_client(BRIDGE_DEACTIVATION_MAX="1") service = scim_bridge.ScimBridgeService(config) status, body = scim("POST", "/scim/v2/Users", self.payload("murky@pingcap.com")) uid = body["id"] FakeNexus.fail_patch[0], FakeNexus.fail_patch[1] = True, 500 try: with self.assertRaises(scim_bridge.BridgeError): service.delete_user(client, uid) finally: FakeNexus.fail_patch[0] = False with self.assertRaises(scim_bridge.BridgeError) as caught: service.delete_user(client, uid) # budget spent -> breaker trips self.assertEqual(caught.exception.status, 429) def test_06z3_cache_honors_server_expires_in(self): import time as _time _, client = self._direct_client() scim_bridge.NexusClient._token_cache.pop(client._cache_key, None) client._token() _, expiry = scim_bridge.NexusClient._token_cache[client._cache_key] remaining = expiry - _time.monotonic() # server said 7200s; cached lifetime must be 7200 minus the margin, # NOT the old 23h assumption self.assertLess(remaining, 7200) self.assertGreater(remaining, 6000) def test_06z4_invalid_put_payload_rejected_not_wiped(self): status, body = scim("POST", "/scim/v2/Users", self.payload("keepme@pingcap.com", title="Lead")) uid = body["id"] # schema-less empty object must be rejected, not treated as full replace status, _ = scim("PUT", f"/scim/v2/Users/{uid}", {}) self.assertEqual(status, 400) self.assertEqual(FakeNexus.users[uid]["Title"], "Lead") # untouched def test_06z5_bootstrap_admin_must_be_configured_and_exempt(self): # exempt list present but no bootstrap admin -> fail closed with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "x@pingcap.com"}) # bootstrap admin not in the exempt list -> fail closed with self.assertRaises(ValueError): scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "x@pingcap.com", "BRIDGE_BOOTSTRAP_ADMIN": "root@pingcap.com"}) # both present and consistent -> ok scim_bridge.BridgeConfig({"BRIDGE_EXEMPT_EMAILS": "root@pingcap.com", "BRIDGE_BOOTSTRAP_ADMIN": "root@pingcap.com"}) # both admin vars set but only one exempt -> fail closed (no override) with self.assertRaises(ValueError): scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "typo@pingcap.com", "BRIDGE_BOOTSTRAP_ADMIN": "typo@pingcap.com", "VERIFY_USERNAME": "real-admin@pingcap.com"}) # both admin vars set and both exempt -> ok scim_bridge.BridgeConfig({ "BRIDGE_EXEMPT_EMAILS": "a1@pingcap.com,a2@pingcap.com", "BRIDGE_BOOTSTRAP_ADMIN": "a1@pingcap.com", "VERIFY_USERNAME": "a2@pingcap.com"}) def test_07_token_cached_once(self): # Valid-credential requests across every test rode the cache: the # bridge minted a real token at most twice. Wrong-secret requests DO # re-hit the token endpoint each time (test_06b), so total calls are # higher -- only successful mints prove caching. handler = self.nexus.RequestHandlerClass self.assertGreaterEqual(handler.token_success[0], 1) # dozens of SCIM calls ran across this suite; a handful of mints # (initial + the two poisoned-cache heal tests) proves caching works self.assertLessEqual(handler.token_success[0], 4) self.assertGreater(handler.token_calls[0], handler.token_success[0]) if __name__ == "__main__": unittest.main(verbosity=2)