"""Prod server-to-server callers must not authenticate as the browser client. The target admits `nexuscrm-ui` on the password grant only for requests carrying browser session-control evidence. A server-to-server caller cannot produce it, so binding that client yields an opaque HTTP 400 -- and, for the cursor-coupled Lead assignment replay, advances the replay queue toward dead_letter on every retry. """ import importlib.util import json import os import re import subprocess import sys import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BROWSER_CLIENT_ID = "nexuscrm-ui" CLIENT_FILE = ROOT / "packs/NOT-EDIT-final-merged-packs/connected_apps/salesforce_compat_client.json" REPLAY_SCRIPT = ROOT / "scripts/salesforce_migration/replay_lead_assignment_from_export.py" RECONCILE_SCRIPT = ROOT / "scripts/org_reconcile/reconcile_org.py" BINDER = ROOT / "scripts/bind_programmatic_oauth_client.py" BOUND_WORKFLOWS = ( ROOT / ".github/workflows/salesforce-prod-lead-campaign-incremental-sync.yml", ROOT / ".github/workflows/prod-org-reconcile.yml", ROOT / ".github/workflows/salesforce-staging-lead-campaign-incremental-replay-sync.yml", ) BINDING_STEP = "Bind installed Salesforce-compatible OAuth client" # Any of these entry points can reach a login that the admission gate judges. # The replay script is reached indirectly: run_hourly_delta_sync.py and # run_hourly_delta_replay_queue.py both spawn it without passing client args. AUTHENTICATING_SCRIPTS = ( "replay_lead_assignment_from_export.py", "run_hourly_delta_replay_queue.py", "run_hourly_delta_catchup.py", "run_hourly_delta_sync.py", "reconcile_org.py", ) def load_replay_module(): spec = importlib.util.spec_from_file_location("replay_lead_assignment", REPLAY_SCRIPT) module = importlib.util.module_from_spec(spec) sys.path.insert(0, str(REPLAY_SCRIPT.parent)) try: spec.loader.exec_module(module) finally: sys.path.pop(0) return module class InstalledConnectedAppTests(unittest.TestCase): def test_pack_ships_a_programmatic_password_grant_client(self): client = json.loads(CLIENT_FILE.read_text(encoding="utf-8")) self.assertEqual(client["sessionIssuerClass"], "salesforce_compat") self.assertTrue(client["isActive"]) self.assertIn("password", client["allowedGrantTypes"]) self.assertNotEqual(client["consumerKey"], BROWSER_CLIENT_ID) for field in ("consumerKey", "consumerSecret"): self.assertIsInstance(client[field], str) self.assertTrue(client[field].strip()) class ReplayClientBindingTests(unittest.TestCase): def setUp(self): self.module = load_replay_module() def test_browser_client_is_refused_before_any_login_attempt(self): with self.assertRaises(ValueError) as caught: self.module.validate_oauth_client(BROWSER_CLIENT_ID, "nexuscrm-ui-internal") self.assertIn("browser session-control", str(caught.exception)) def test_missing_client_fails_closed_rather_than_defaulting(self): for client_id, client_secret in (("", "secret"), ("client", ""), ("", ""), (" ", " ")): with self.subTest(client_id=client_id, client_secret=client_secret): with self.assertRaises(ValueError): self.module.validate_oauth_client(client_id, client_secret) def test_programmatic_client_is_accepted(self): client = json.loads(CLIENT_FILE.read_text(encoding="utf-8")) self.module.validate_oauth_client(client["consumerKey"], client["consumerSecret"]) def test_argparse_default_comes_from_the_environment_not_the_browser_client(self): source = REPLAY_SCRIPT.read_text(encoding="utf-8") self.assertNotIn('default="nexuscrm-ui"', source) self.assertNotIn('default="nexuscrm-ui-internal"', source) self.assertIn('os.environ.get("NEXUSCRM_OAUTH_CLIENT_ID", "")', source) self.assertIn('os.environ.get("NEXUSCRM_OAUTH_CLIENT_SECRET", "")', source) class ReconcileClientBindingTests(unittest.TestCase): def test_client_is_not_hardcoded(self): source = RECONCILE_SCRIPT.read_text(encoding="utf-8") self.assertNotIn('OAUTH_CLIENT_ID = "nexuscrm-ui"', source) self.assertIn('os.environ.get("NEXUSCRM_OAUTH_CLIENT_ID", "")', source) self.assertIn('os.environ.get("NEXUSCRM_OAUTH_CLIENT_SECRET", "")', source) def test_rate_limiting_still_retries_while_credential_errors_stop(self): # Loaded by source rather than import: the module authenticates at import # time, so importing it here would attempt a live token request. source = RECONCILE_SCRIPT.read_text(encoding="utf-8") namespace = {} exec( # noqa: S102 - executing the repo's own reviewed source, not input compile( re.search( r"^RETRYABLE_HTTP_STATUSES.*?^def token\(\):", source, re.MULTILINE | re.DOTALL, ).group(0).rsplit("def token():", 1)[0], str(RECONCILE_SCRIPT), "exec", ), namespace, ) retryable = namespace["token_status_is_retryable"] # Transient: the hourly API quota clears on its own. for code in (408, 429, 500, 502, 503, 504): with self.subTest(code=code): self.assertTrue(retryable(code), f"HTTP {code} must still retry") # Deterministic: no number of retries turns these into a token. for code in (400, 401, 403, 404): with self.subTest(code=code): self.assertFalse(retryable(code), f"HTTP {code} must stop early") def run_reconcile(self, env_overrides): env = dict(os.environ) env.update( PROD_URL="https://target.invalid", PROD_USERNAME="admin@example.invalid", PROD_PASSWORD="unused-by-this-assertion", ) env.update(env_overrides) return subprocess.run( [sys.executable, str(RECONCILE_SCRIPT)], env=env, capture_output=True, text=True, cwd=str(ROOT), timeout=60, ) def test_missing_client_env_fails_before_any_network_call(self): result = self.run_reconcile({ "NEXUSCRM_OAUTH_CLIENT_ID": "", "NEXUSCRM_OAUTH_CLIENT_SECRET": "", }) self.assertNotEqual(result.returncode, 0) self.assertIn("NEXUSCRM_OAUTH_CLIENT_ID", result.stdout + result.stderr) def test_browser_client_env_is_refused(self): result = self.run_reconcile({ "NEXUSCRM_OAUTH_CLIENT_ID": BROWSER_CLIENT_ID, "NEXUSCRM_OAUTH_CLIENT_SECRET": "nexuscrm-ui-internal", }) self.assertNotEqual(result.returncode, 0) self.assertIn("browser session-control", result.stdout + result.stderr) class BindingScriptTests(unittest.TestCase): def run_binder(self, tmp_env_path, cwd=ROOT): env = dict(os.environ, GITHUB_ENV=str(tmp_env_path)) return subprocess.run( [sys.executable, str(BINDER)], env=env, capture_output=True, text=True, cwd=str(cwd), timeout=60, ) def test_binds_the_installed_client_and_masks_the_secret(self): import tempfile with tempfile.TemporaryDirectory() as tmp: env_file = Path(tmp) / "github_env" env_file.touch() result = self.run_binder(env_file) self.assertEqual(result.returncode, 0, result.stderr) client = json.loads(CLIENT_FILE.read_text(encoding="utf-8")) written = env_file.read_text(encoding="utf-8") self.assertIn(f"NEXUSCRM_OAUTH_CLIENT_ID={client['consumerKey']}\n", written) self.assertIn(f"NEXUSCRM_OAUTH_CLIENT_SECRET={client['consumerSecret']}\n", written) # The mask must reach the step log, and the secret must not. self.assertIn(f"::add-mask::{client['consumerSecret']}", result.stdout) secret_lines = [ line for line in result.stdout.splitlines() if client["consumerSecret"] in line and not line.startswith("::add-mask::") ] self.assertEqual(secret_lines, []) def test_refuses_to_bind_when_metadata_is_missing(self): import tempfile with tempfile.TemporaryDirectory() as tmp: env_file = Path(tmp) / "github_env" env_file.touch() result = self.run_binder(env_file, cwd=tmp) self.assertNotEqual(result.returncode, 0) self.assertIn("missing", result.stdout + result.stderr) self.assertEqual(env_file.read_text(encoding="utf-8"), "") def test_refuses_to_run_outside_a_workflow(self): env = {k: v for k, v in os.environ.items() if k != "GITHUB_ENV"} result = subprocess.run( [sys.executable, str(BINDER)], env=env, capture_output=True, text=True, cwd=str(ROOT), timeout=60, ) self.assertNotEqual(result.returncode, 0) self.assertIn("GITHUB_ENV", result.stdout + result.stderr) class WorkflowCoverageTests(unittest.TestCase): """Derive the set that needs binding from the call graph, not a hand list. The first round of this change bound only the two prod workflows and missed the staging replay workflow, which reaches the same replay script indirectly through run_hourly_delta_sync.py. A hardcoded list cannot catch that class of miss; enumerating callers can. """ def workflows_reaching_a_login(self): reaching = {} for workflow in sorted((ROOT / ".github/workflows").glob("*.yml")): text = workflow.read_text(encoding="utf-8") hits = [name for name in AUTHENTICATING_SCRIPTS if name in text] if hits: reaching[workflow] = (text, hits) return reaching def test_every_workflow_that_can_log_in_binds_a_programmatic_client(self): reaching = self.workflows_reaching_a_login() self.assertTrue(reaching, "call-graph scan found nothing; the probe itself is broken") for workflow, (text, hits) in reaching.items(): with self.subTest(workflow=workflow.name, via=hits): if "--skip-replay" in text and "reconcile_org.py" not in text: # Never spawns the replay script, so it never logs in. self.assertNotIn(BINDING_STEP, text) continue self.assertIn( BINDING_STEP, text, f"{workflow.name} can reach a login via {hits} but binds no client", ) def test_the_declared_bound_set_matches_the_call_graph(self): expected = { workflow for workflow, (text, _) in self.workflows_reaching_a_login().items() if not ("--skip-replay" in text and "reconcile_org.py" not in text) } self.assertEqual(expected, set(BOUND_WORKFLOWS)) class WorkflowBindingTests(unittest.TestCase): def test_prod_workflows_call_the_binder(self): for workflow in BOUND_WORKFLOWS: with self.subTest(workflow=workflow.name): text = workflow.read_text(encoding="utf-8") self.assertIn("Bind installed Salesforce-compatible OAuth client", text) self.assertIn("python3 scripts/bind_programmatic_oauth_client.py", text) # jq is not guaranteed on either self-hosted runner. self.assertNotIn("jq ", text) def test_binding_runs_before_the_step_that_authenticates(self): sync = BOUND_WORKFLOWS[0].read_text(encoding="utf-8") self.assertLess( sync.index("Bind installed Salesforce-compatible OAuth client"), sync.index("Run update-or-create catch-up sync"), ) reconcile = BOUND_WORKFLOWS[1].read_text(encoding="utf-8") self.assertLess( reconcile.index("Bind installed Salesforce-compatible OAuth client"), reconcile.index("Run Org reconcile"), ) def test_no_prod_workflow_still_passes_the_browser_client(self): for workflow in BOUND_WORKFLOWS: with self.subTest(workflow=workflow.name): text = workflow.read_text(encoding="utf-8") offending = [ line for line in text.splitlines() # The binding step's own guard names the browser client in a # jq inequality; that is the assertion, not a use of it. if BROWSER_CLIENT_ID in line and not re.search(r'\.\s*!=\s*"nexuscrm-ui"', line) ] self.assertEqual(offending, []) if __name__ == "__main__": unittest.main()