import hashlib import json import os import pathlib import re import shutil import subprocess import tempfile import unittest ROOT = pathlib.Path(__file__).resolve().parents[1] MODULE = ROOT / "appservices" / "sfcrm" TRUSTED_PUBLISHER = ROOT / ".github" / "workflows" / "trusted-publish-sfcrm-appservice-image.yml" PUBLISHER_CALLER = ROOT / ".github" / "workflows" / "publish-sfcrm-appservice-image.yml" SIGNING_POLICY = ROOT / "release" / "sfcrm-appservice-image-signing-policy.json" SIGNING_PUBLIC_KEY = ROOT / "release" / "sfcrm-appservice-image-signing-public-key.pem" class SfcrmAppServiceBoundaryTest(unittest.TestCase): def test_record_type_display_parity_fixtures_match(self): appservice_fixture = ( MODULE / "testdata" / "record-type-display-parity.json" ) dialect_fixture = ( ROOT / "dialects" / "dialect-sf-crm" / "testdata" / "record-type-display-parity.json" ) self.assertEqual( appservice_fixture.read_bytes(), dialect_fixture.read_bytes(), "independently published modules must enforce the same RecordType display obligation", ) payload = json.loads(appservice_fixture.read_text()) self.assertEqual( { "objectApiName", "recordId", "recordTypeField", "relationshipField", "recordTypeId", "recordTypeName", }, set(payload), ) for field in payload: self.assertTrue(payload[field], field) def test_local_branch_image_builds_with_exact_sibling_workspace_only(self): script = ROOT / "scripts" / "build-d3-sfcrm-dev-image.sh" with tempfile.TemporaryDirectory() as temporary: temporary_root = pathlib.Path(temporary) repository = temporary_root / "app-packs" (repository / "scripts").mkdir(parents=True) shutil.copy2(script, repository / "scripts" / script.name) for module in ( "appservices/sfcrm", "sfcrm/querycodec", "sfcrm/scim", "sfcrm/wire", ): (repository / module).mkdir(parents=True) subprocess.run(["git", "init", "--quiet"], cwd=repository, check=True) subprocess.run(["git", "add", "."], cwd=repository, check=True) subprocess.run( [ "git", "-c", "user.name=NexusCRM Test", "-c", "user.email=nexuscrm-test@example.invalid", "commit", "--quiet", "-m", "fixture", ], cwd=repository, check=True, ) fake_bin = temporary_root / "bin" fake_bin.mkdir() go_log = temporary_root / "go.log" (fake_bin / "go").write_text( "#!/usr/bin/env bash\n" "printf '%s|%s|%s\\n' \"${GOWORK-}\" \"$PWD\" \"$*\" >> \"$FAKE_GO_LOG\"\n" "if [[ \"${1-}\" == work && \"${2-}\" == init ]]; then : > go.work; exit 0; fi\n" "if [[ \"${1-}\" == build ]]; then exit 79; fi\n" "exit 0\n", ) for command in ("go", "docker", "jq", "gh"): path = fake_bin / command if command != "go": path.write_text("#!/usr/bin/env bash\nexit 0\n") path.chmod(0o755) environment = { **os.environ, "PATH": f"{fake_bin}{os.pathsep}{os.environ['PATH']}", "FAKE_GO_LOG": str(go_log), } local = subprocess.run( [ str(repository / "scripts" / script.name), "--local-only", "--tag-prefix", "nexuscrm-test/test1/branch-run", "--output-manifest", str(temporary_root / "manifest.json"), ], cwd=repository, env=environment, capture_output=True, text=True, ) self.assertEqual(79, local.returncode, local.stderr) local_invocations = go_log.read_text().splitlines() self.assertEqual(2, len(local_invocations), local_invocations) workspace_env, workspace_cwd, workspace_command = local_invocations[0].split("|", 2) self.assertEqual("off", workspace_env) self.assertNotEqual(str(repository), workspace_cwd) self.assertEqual( "work init " + " ".join( str(repository / module) for module in ( "appservices/sfcrm", "sfcrm/querycodec", "sfcrm/scim", "sfcrm/wire", ) ), workspace_command, ) build_workspace, build_cwd, build_command = local_invocations[1].split("|", 2) self.assertTrue(build_workspace.endswith("/go.work"), build_workspace) self.assertNotEqual("off", build_workspace) self.assertEqual(str(repository / "appservices" / "sfcrm"), build_cwd) self.assertTrue(build_command.startswith("build -trimpath "), build_command) go_log.unlink() published = subprocess.run( [str(repository / "scripts" / script.name)], cwd=repository, env=environment, capture_output=True, text=True, ) self.assertEqual(79, published.returncode, published.stderr) publish_invocations = go_log.read_text().splitlines() self.assertEqual(1, len(publish_invocations), publish_invocations) publish_workspace, publish_cwd, publish_command = publish_invocations[0].split("|", 2) self.assertEqual("off", publish_workspace) self.assertEqual(str(repository / "appservices" / "sfcrm"), publish_cwd) self.assertTrue(publish_command.startswith("build -trimpath "), publish_command) def test_production_code_composes_public_l0_owners(self): production = "\n".join( path.read_text() for path in MODULE.rglob("*.go") if not path.name.endswith("_test.go") ) self.assertIn("processhttp.NewProcess", production) self.assertIn("runtimegrpc.NewRuntimeClientGroup", production) self.assertIn("runtimeClients.DialRecordAccessRuntime", production) for forbidden in ( '"crypto/tls"', '"crypto/x509"', '"database/sql"', '"google.golang.org/grpc"', '"google.golang.org/grpc/credentials"', 'grpc.NewClient', 'net.Listen', 'net.Dial', 'net.Dialer', 'net.ListenConfig', 'http.Server{', 'tls.Config{', ): self.assertNotIn(forbidden, production) def test_module_uses_released_dependencies_without_replace(self): go_mod = (MODULE / "go.mod").read_text() self.assertNotRegex(go_mod, re.compile(r"(?m)^\s*replace(?:\s|\()")) self.assertRegex(go_mod, r"github.com/pingcap-inc/nexuscrm-api-host v\d+\.\d+\.\d+") self.assertRegex(go_mod, r"github.com/pingcap-inc/nexuscrm-contract v\d+\.\d+\.\d+") codec_module = "github.com/pingcap-inc/nexuscrm-app-packs/sfcrm/querycodec" consumers = ( MODULE / "go.mod", ROOT / "dialects" / "dialect-sf-crm" / "go.mod", ROOT / "dialects" / "dialect-sf-crm-soql" / "go.mod", ) versions = [] for consumer in consumers: contents = consumer.read_text() versions_in_manifest = re.findall( rf"(?m)^\s*(?:require\s+)?{re.escape(codec_module)}\s+(v\S+)$", contents, ) self.assertEqual(len(versions_in_manifest), 1, consumer) self.assertNotRegex( contents, re.compile(rf"(?m)^\s*(?:replace\s+)?{re.escape(codec_module)}(?:\s+v\S+)?\s+=>"), ) versions.append(versions_in_manifest[0]) self.assertEqual(versions, ["v0.1.6-exp.query-field-date.2"] * len(consumers)) def test_query_family_has_one_product_grammar_owner(self): codec = (ROOT / "sfcrm" / "querycodec" / "handler.go").read_text() self.assertIn("func (h *Handler) execute", codec) self.assertIn("func (h *Handler) queryMore", codec) dialect_query = (ROOT / "dialects" / "dialect-sf-crm" / "query.go").read_text() dialect_history = (ROOT / "dialects" / "dialect-sf-crm" / "query_history.go").read_text() adapter = (ROOT / "dialects" / "dialect-sf-crm" / "query_codec_adapter.go").read_text() self.assertIn("serveQueryWithCodec", dialect_query) self.assertIn("querycodec.NewHandler", adapter) for stale_owner in ( "ParseSOQL", "decodeQueryLocator", "formatQueryResponse", "applySOQLResultWindow", "executeHistoryQueryWithSOQL", "respondHistoryQueryError", ): self.assertNotIn(stale_owner, dialect_query + dialect_history) def test_mutation_projection_uses_the_public_l1_wire_owner(self): production = "\n".join( path.read_text() for path in MODULE.rglob("*.go") if not path.name.endswith("_test.go") ) self.assertIn( '"github.com/pingcap-inc/nexuscrm-app-packs/sfcrm/wire"', production, ) for forbidden in ( '"github.com/nexuscrm/dialect-sf-crm', '"github.com/nexuscrm/shared', '"github.com/pingcap-inc/NexusCRM', ): self.assertNotIn(forbidden, production) def test_production_composition_matches_latest_immutable_app_release(self): command = (MODULE / "cmd" / "sfcrm-appservice" / "main.go").read_text() appservice = "\n".join( path.read_text() for path in MODULE.glob("*.go") if not path.name.endswith("_test.go") ) constants = dict( re.findall(r'(\w*CapabilityID)\s*=\s*"([^"]+)"', appservice) ) google_login_state_capability = "io.nexuscrm.sfcrm.google-login-state.v1" self.assertEqual( constants["GoogleLoginStateCapabilityID"], google_login_state_capability, ) advertised_match = re.search( r"func processCapabilities\(\) \[\]string \{\s*return \[\]string\{([^}]*)\}", command, re.DOTALL, ) self.assertIsNotNone(advertised_match) self.assertRegex(command, r"Capabilities:\s*capabilities\b") advertised = { constants[name] for name in re.findall( r"appservice\.(\w*CapabilityID)", advertised_match.group(1) ) } release_root = ROOT / "releases" / "io.nexuscrm.sfcrm" releases = [ (tuple(int(part) for part in path.name.split(".")), path) for path in release_root.iterdir() if path.is_dir() and re.fullmatch(r"\d+\.\d+\.\d+", path.name) ] _, latest_release = max(releases) artifact = json.loads((latest_release / "app-package.json").read_text()) released = { capability["capability_id"] for capability in artifact["canonical_manifest"]["provided_capabilities"] } required_runtime = set( artifact["canonical_manifest"]["required_engine_capabilities"] ) consumer_only = set() if "runtime.identity.session-lifecycle.v1" in required_runtime: consumer_only.update({ constants["SessionCapabilityID"], constants["GoogleLoginStateCapabilityID"], }) installed = released | consumer_only self.assertEqual(advertised, installed) self.assertNotIn(constants["SessionCapabilityID"], released) dockerfile = (MODULE / "Dockerfile").read_text() self.assertIn( f'io.nexuscrm.service-capabilities="[\\"{constants["GoogleLoginStateCapabilityID"]}\\"]"', dockerfile, ) for workflow_path in ( ROOT / ".github" / "workflows" / "sfcrm-appservice.yml", TRUSTED_PUBLISHER, ): self.assertIn( f"'[\"{google_login_state_capability}\"]'", workflow_path.read_text(), ) oauth_callback = ( ROOT / "salesforce-crm-standard" / "frontend" / "src" / "platform" / "components" / "auth" / "OAuthCallback.tsx" ).read_text() self.assertIn( "actions.exchangeOAuthCode(provider, code, savedState)", oauth_callback, ) frontend_auth = ( ROOT / "salesforce-crm-standard" / "frontend" / "src" / "platform" / "infrastructure" / "api" / "auth.ts" ).read_text() frontend_endpoints = ( ROOT / "salesforce-crm-standard" / "frontend" / "src" / "platform" / "infrastructure" / "api" / "endpoints.ts" ).read_text() self.assertNotIn("buildOAuthState('google')", frontend_auth) self.assertIn("await this.beginGoogleLogin(redirectUri)", frontend_auth) self.assertIn("GOOGLE_LOGIN_STATE: '/services/oauth2/google/login-state'", frontend_endpoints) self.assertNotIn("SearchScopeOrderCapabilityID", constants) self.assertNotIn("io.nexuscrm.sfcrm.search-scope-order.v1", released) self.assertNotIn("catalog.consumer.app-metadata-view.v1", required_runtime) self.assertNotIn("RuntimeGetAppMetadataView:", command) wiring = { constants["CapabilityID"]: ( "DialRecordAccessRuntime", "RuntimeEvaluate:", ), constants["QueryCapabilityID"]: ( "DialQueryRuntime", "RuntimeQuery:", ), constants["MutationCapabilityID"]: ( "DialRecordMutationRuntime", "RuntimeMutate:", ), constants["RelatedCountCapabilityID"]: ( "DialRelatedCountRuntime", "RuntimeRelatedCount:", "relatedCountRuntime.Ready()", "relatedCountRuntime.Metrics()", "relatedCountRuntime.Close()", ), constants["RecycleBinCapabilityID"]: ( "DialRecycleBinRuntime", "RuntimeRecycleBin:", "recycleBinRuntime.Ready()", "recycleBinRuntime.Metrics()", "recycleBinRuntime.Close()", ), constants["RecycleBinMutationCapabilityID"]: ( "DialRecycleBinRuntime", "RuntimeRestoreRecycleBin:", "RuntimePurgeRecycleBin:", "recycleBinRuntime.Ready()", "recycleBinRuntime.Metrics()", "recycleBinRuntime.Close()", ), constants["RecordViewCapabilityID"]: ( "DialRecordViewProofRuntime", "RuntimeReadRecordView:", "recordViewRuntime.Ready()", "recordViewRuntime.Metrics()", "recordViewRuntime.Close()", ), constants["RecordViewAttachmentSourceCapabilityID"]: ( "DialRecordViewProofRuntime", "RuntimeResolveAttachmentSourceProofs:", ), constants["EffectivePermissionsCapabilityID"]: ( "RuntimeReadEffectivePermissions:", "platformRuntime.ReadEffectivePermissions", ), constants["SearchCapabilityID"]: ( "RuntimeReadSearchScopeOrder:", "platformRuntime.ReadSearchScopeOrder", "RuntimeReadEffectivePermissions:", "platformRuntime.ReadEffectivePermissions", "RuntimeGetAppMetadataViewV3:", "platformRuntime.GetAppMetadataViewV3", "RuntimeSearch:", "platformRuntime.Search", ), constants["FlowActionsCapabilityID"]: ( "RuntimeGetAppMetadataViewV3:", "platformRuntime.GetAppMetadataViewV3", "RuntimeReadEffectivePermissions:", "platformRuntime.ReadEffectivePermissions", "RuntimeExecuteFlow:", "platformRuntime.ExecuteFlow", "RuntimeExecuteCatalogAction:", "platformRuntime.ExecuteCatalogAction", "RuntimePreviewCatalogAction:", "platformRuntime.PreviewCatalogAction", ), constants["ApprovalLifecycleCapabilityID"]: ( "RuntimeExecuteApproval:", "platformRuntime.ExecuteApproval", ), constants["ApprovalReadCapabilityID"]: ( "RuntimeReadApproval:", "platformRuntime.ReadApproval", ), constants["ApprovalNotificationFlowsCapabilityID"]: ( "RuntimeReadNotificationFlows:", "platformRuntime.ReadApprovalNotificationFlows", ), constants["SessionCapabilityID"]: ( "RuntimeCreateSession:", "platformRuntime.CreateSession", "RuntimeReadSession:", "platformRuntime.ReadSession", "RuntimeRevokeSession:", "platformRuntime.RevokeSession", ), } for capability, required_tokens in wiring.items(): for token in required_tokens: self.assertEqual( token in command, capability in installed, f"{token} disagrees with {capability} in {latest_release.name}", ) @unittest.skipUnless( os.environ.get("NEXUSCRM_VERIFY_SFCRM_DEPENDENCY_GRAPH") == "1", "full private-module graph is verified by the trusted exact-main job", ) def test_mutation_production_dependency_graph_has_no_core_or_dialect_leaks(self): dependencies = subprocess.run( ["go", "list", "-deps", "-test", "./..."], cwd=MODULE, env={**os.environ, "GOWORK": "off"}, check=True, capture_output=True, text=True, ).stdout.splitlines() forbidden_dependencies = { dependency for dependency in dependencies if dependency == "github.com/nexuscrm/backend" or dependency.startswith("github.com/nexuscrm/backend/") or dependency == "github.com/nexuscrm/shared" or dependency.startswith("github.com/nexuscrm/shared/") or dependency == "github.com/nexuscrm/dialect-sf-crm" or dependency.startswith("github.com/nexuscrm/dialect-sf-crm/") } self.assertEqual(forbidden_dependencies, set()) def test_no_second_dormant_record_access_client_remains(self): owners = [ path for path in ROOT.rglob("*.go") if "type RuntimeRecordAccessClient struct" in path.read_text() ] self.assertEqual(owners, [MODULE / "consumer.go"]) def test_exact_main_validation_has_an_explicit_sha_dispatch(self): workflow = (ROOT / ".github" / "workflows" / "sfcrm-appservice.yml").read_text() self.assertNotIn("workflow_dispatch:", workflow) self.assertIn("repository_dispatch:", workflow) self.assertIn("types: [sfcrm-appservice-exact-main]", workflow) self.assertIn("github.event.client_payload.sha", workflow) self.assertIn("supplied !== commit.sha", workflow) self.assertIn("ref: ${{ steps.source.outputs.sha }}", workflow) self.assertIn('NEXUSCRM_VERIFY_SFCRM_DEPENDENCY_GRAPH: "1"', workflow) def test_appservice_publisher_activates_the_dependency_blind_signer(self): self.assertTrue(TRUSTED_PUBLISHER.is_file()) self.assertTrue(PUBLISHER_CALLER.is_file()) policy = json.loads(SIGNING_POLICY.read_text()) trusted_ref = ( "pingcap-inc/nexuscrm-app-packs/.github/workflows/" "trusted-publish-sfcrm-appservice-image.yml@ad6eba21a639aa1698f9da069776757e38d1b107" ) self.assertEqual(policy["publisher_state"], "active") self.assertEqual(policy["trusted_workflow_ref"], trusted_ref) self.assertEqual(policy["owner_repository"], "pingcap-inc/nexuscrm-app-packs") self.assertEqual(policy["repository_id"], "1236725461") self.assertEqual(policy["image_repository"], "ghcr.io/pingcap-inc/nexuscrm-sfcrm-appservice") self.assertEqual( policy["public_key_sha256"], "sha256:" + hashlib.sha256(SIGNING_PUBLIC_KEY.read_bytes()).hexdigest(), ) key = subprocess.run( ["openssl", "pkey", "-pubin", "-in", str(SIGNING_PUBLIC_KEY), "-text", "-noout"], check=True, capture_output=True, text=True, ).stdout self.assertIn("ASN1 OID: prime256v1", key) caller = PUBLISHER_CALLER.read_text() self.assertIn("workflow_run:", caller) self.assertIn("workflows: [sfcrm-appservice]", caller) self.assertIn('gh run download "$RUN_ID"', caller) self.assertIn("appservice-publication-receipt-$RUN_ATTEMPT", caller) self.assertIn(f"uses: {trusted_ref}", caller) self.assertIn("source_sha: ${{ needs.receipt.outputs.target_sha }}", caller) publish_call = caller.split("\n publish:\n", 1)[1] self.assertNotIn("\n secrets:\n", publish_call) for permission in ( "actions: read", "contents: read", "packages: write", "id-token: write", ): self.assertIn(permission, publish_call) self.assertNotIn("actions: write", publish_call) self.assertNotIn("contents: write", publish_call) self.assertNotIn("github.event.workflow_run.head_sha", caller) self.assertNotIn("workflow_dispatch:", caller) self.assertNotIn("pull_request:", caller) self.assertNotIn("actions/attest", caller) workflow = TRUSTED_PUBLISHER.read_text() self.assertIn("workflow_call:", workflow) self.assertNotIn("workflow_dispatch:", workflow) self.assertNotIn("repository_dispatch:", workflow) self.assertNotIn("actions/attest", workflow) self.assertIn("test \"$GITHUB_EVENT_NAME\" = workflow_run", workflow) self.assertIn('test "${{ github.event.workflow_run.name }}" = sfcrm-appservice', workflow) self.assertIn('gh run download "$RUN_ID"', workflow) self.assertIn("appservice-publication-receipt-$RUN_ATTEMPT", workflow) self.assertIn("sfcrm-appservice-binary-$RUN_ATTEMPT", workflow) self.assertIn("nexuscrm.appservice-binary/v1", workflow) self.assertIn("actual_binary_sha256", workflow) self.assertIn(".target_sha == $source_sha", workflow) self.assertNotIn("github.event.workflow_run.head_sha", workflow) self.assertEqual(workflow.count("id-token: write"), 1) self.assertIn("appservices/sfcrm/Dockerfile", workflow) self.assertIn("if: ${{ steps.existing.outputs.exists != 'true' }}", workflow) self.assertIn("if: ${{ needs.candidate.outputs.exists == 'true' }}", workflow) self.assertNotIn("merge-base --is-ancestor", workflow) self.assertGreaterEqual(workflow.count('test "$current_main" = "$SOURCE_SHA"'), 2) self.assertIn('mkdir -p "$image_context/appservices/sfcrm"', workflow) self.assertIn('install -m 0755 "$binary"', workflow) self.assertIn('test "$current_main" = "$SOURCE_SHA"', workflow) for forbidden in ( "actions/setup-go", "go mod download", "go build", "NEXUSCRM_API_HOST_DEPLOY_KEY", "NEXUSCRM_CONTRACT_DEPLOY_KEY", "NEXUSCRM_APP_PACKS_TOKEN", ): self.assertNotIn(forbidden, workflow) appservice_ci = (ROOT / ".github" / "workflows" / "sfcrm-appservice.yml").read_text() self.assertIn("Validate publisher Docker context shape", appservice_ci) self.assertIn('mkdir -p "$context/appservices/sfcrm"', appservice_ci) self.assertIn("Write resolver-owned AppService publication receipt", appservice_ci) self.assertIn("nexuscrm.appservice-publication-receipt/v1", appservice_ci) self.assertIn("name: appservice-publication-receipt-${{ github.run_attempt }}", appservice_ci) self.assertIn("nexuscrm.appservice-binary/v1", appservice_ci) self.assertIn("name: sfcrm-appservice-binary-${{ github.run_attempt }}", appservice_ci) self.assertIn('-o "$image_context/appservices/sfcrm/sfcrm-appservice-binary"', appservice_ci) self.assertIn(".github/workflows/publish-sfcrm-appservice-image.yml", appservice_ci) self.assertIn(".github/workflows/trusted-publish-sfcrm-appservice-image.yml", appservice_ci) self.assertIn("NEXUSCRM_APP_PACKS_TOKEN: ${{ github.token }}", appservice_ci) self.assertIn("http.https://github.com/.extraheader", appservice_ci) self.assertIn("x-access-token:%s", appservice_ci) self.assertIn("persist-credentials: false", appservice_ci) self.assertNotIn("git config --global http.https://github.com/.extraheader", appservice_ci) action_refs = re.findall(r"uses:\s+(actions/[^@\s]+)@([^\s#]+)", appservice_ci) self.assertTrue(action_refs) for action, revision in action_refs: self.assertRegex(revision, r"^[0-9a-f]{40}$", action) if __name__ == "__main__": unittest.main()