#!/usr/bin/env python3 """Collect declared release references through the official CLI; never runs business calls.""" import argparse import datetime import json from pathlib import Path import re import subprocess import tempfile def cli(*args): result = subprocess.run(['baijimu', '--json', *map(str, args)], capture_output=True, text=True) if result.returncode: raise ValueError('Official CLI query failed; inspect the same command locally') reply = json.loads(result.stdout) if reply.get('errorCode') != '0': raise ValueError('Owner returned a failed CModel response') return reply['data'] def require(condition, message): if not condition: raise ValueError(message) def exact_version(value): require(isinstance(value, str) and re.fullmatch( r'(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)' r'(?:-(?:0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)' r'(?:\.(?:0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*)?' r'(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?', value), 'Exact SemVer required') return value def one(items, predicate, message): found = [item for item in items if predicate(item)] require(len(found) == 1, message) return found[0] def bundle_evidence(spec, query): workspace, bundle, version = spec['workspaceId'], spec['bundleId'], exact_version(spec['version']) selected = one(query('bundle', 'version', 'list', bundle, '--workspace-id', workspace), lambda item: item['semanticVersion'] == version, 'Bundle version is absent or ambiguous') require(selected['status'] == 'PUBLISHED', 'Bundle version is not available') commit = selected['sourceGitCommitId'] require(re.fullmatch('[0-9a-f]{40}', commit), 'Bundle source commit is invalid') source = subprocess.run(['git', '-C', spec['projectDirectory'], 'show', commit + ':baijimu.bundle.json'], capture_output=True, text=True) require(source.returncode == 0, 'Exact Bundle source commit is unavailable locally') manifest = json.loads(source.stdout) with tempfile.TemporaryDirectory(prefix='bundle-delivery-') as directory: path = Path(directory) / 'baijimu.bundle.json' path.write_text(source.stdout) checked = query('bundle', 'manifest', 'preflight', bundle, '--workspace-id', workspace, '--file', path) require(checked.get('valid') is True and checked.get('bundleId') == bundle, 'Bundle preflight identity or validation failed') return {'bundleId': bundle, 'version': version, 'sourceProjectId': selected['sourceProjectId'], 'sourceCommitId': commit, 'resources': manifest['definition']['resources'], 'validationScope': checked['validationScope']} def hosted_evidence(spec, query): workspace, project, version = spec['workspaceId'], spec['projectId'], exact_version(spec['version']) source = query('project', 'version', 'get', project, '--workspace-id', workspace, '--version', version) artifact = query('rust-build', 'artifact', 'get', spec['artifactId'], '--workspace-id', workspace) require(source['projectId'] == project and source['workspaceId'] == workspace and source['version'] == version, 'Project version identity mismatch') require(artifact['projectId'] == project and artifact['workspaceId'] == workspace and artifact['id'] == spec['artifactId'] and artifact['version'] == version and artifact['sourceCommitId'] == source['sourceCommitId'] and artifact['status'] == 'READY', 'Artifact does not match the exact project version') environment = one(query('hosted-service', 'env', 'list', '--workspace-id', workspace, '--project-id', project), lambda row: row['id'] == spec['environmentId'], 'Hosted Environment is absent or ambiguous') # The current Environment association, rather than the first historical success, selects deployment. require(environment['projectId'] == project and environment['id'] == spec['environmentId'] and environment['status'] == 'ACTIVE', 'Hosted Environment identity or state mismatch') history = query('hosted-service', 'project', 'deployments', project, '--workspace-id', workspace, '--environment-id', spec['environmentId']) deployment = one(history['deployments'], lambda row: row['id'] == environment['currentDeploymentId'], 'Current deployment is absent or ambiguous in the Owner result') require(deployment['projectId'] == project and deployment['environmentId'] == spec['environmentId'] and deployment['projectVersion'] == version and deployment['status'] == 'SUCCEEDED', 'Current deployment does not match the expected successful version') require(all(row['status'] == 'SUCCEEDED' for row in deployment['databaseMigrations']) and bool(deployment['endpoints']) and all(row['status'] == 'SUCCEEDED' and row['deploymentId'] == deployment['id'] for row in deployment['endpoints']), 'Migration or Endpoint verification is incomplete') return {'workspaceId': workspace, 'projectId': project, 'version': version, 'sourceCommitId': source['sourceCommitId'], 'artifactId': artifact['id'], 'environmentId': environment['id'], 'deploymentId': deployment['id'], 'artifactDeploymentCorrelation': 'IMMUTABLE_PROJECT_VERSION_AND_SOURCE_COMMIT', 'status': 'SUCCEEDED'} def plugin_evidence(spec, query): plugin = query('lifecycle-plugin', 'get', spec['pluginId'], '--workspace-id', spec['workspaceId']) require(isinstance(plugin, dict) and plugin.get('pluginId') == spec['pluginId'] and type(plugin.get('isDeleted')) is int and plugin['isDeleted'] == 0, 'Plugin directory identity is absent or inactive') execution = plugin.get('bundleExecution') require(isinstance(execution, dict) and set(execution) == {'protocols', 'executeUrl', 'supportedStages'}, 'Plugin registration does not use the current declared protocol format') field_map = {'lifecycle': 'lifecycleProtocol', 'reference': 'referenceProtocol', 'credentials': 'credentialsProtocol', 'bindingRead': 'bindingReadProtocol', 'inventory': 'inventoryProtocol', 'propertyWrite': 'propertyWriteProtocol'} expected = {key: exact_version(spec[field]) for key, field in field_map.items()} registered = execution['protocols'] require(isinstance(registered, dict) and set(registered) == set(field_map), 'Plugin registered protocol declaration is missing or incomplete') require({key: exact_version(value) for key, value in registered.items()} == expected, 'Plugin registered protocols do not match the declared exact versions') stages = execution['supportedStages'] require(isinstance(stages, list) and len(stages) == 2 and set(stages) == {'BEFORE_RESOURCES', 'AFTER_RESOURCES'}, 'Plugin resource stages are incomplete or ambiguous') return {'pluginId': plugin['pluginId'], 'protocols': registered, 'registrationVerification': 'MATCHED_DECLARED_PROTOCOLS', 'programVerification': 'NOT_EXECUTED', 'businessVerification': 'NOT_EXECUTED', 'supportedStages': stages} def runtime_evidence(spec, bundle, query): page = query('bundle', 'resources', bundle['bundleId'], '--environment-key', spec['environmentKey'], '--workspace-id', spec['workspaceId'], '--application-runtime-id', spec['applicationRuntimeId'], '--all') require(page['workspaceId'] == spec['workspaceId'] and page['applicationRuntimeId'] == spec['applicationRuntimeId'] and not page['hasMore'], 'Owner resource query scope or pagination mismatch') expected = {f"{spec['environmentKey']}/{bundle['bundleId']}/{row['resourceType']}/{row['resourceKey']}": row['semanticVersion'] for row in bundle['resources']} actual = {row['resourceLocator']: row['semanticVersion'] for row in page['items']} require(len(actual) == len(page['items']) and expected == actual, 'Owner resources differ from the declared exact Bundle scope') return {'applicationRuntimeId': page['applicationRuntimeId'], 'resources': [{'resourceLocator': row['resourceLocator'], 'semanticVersion': row['semanticVersion']} for row in page['items']]} def collect(spec, query=cli): require(set(spec) <= {'bundle', 'hostedServices', 'plugins', 'runtime'} and all(key in spec for key in ['bundle', 'hostedServices', 'plugins']), 'Declare bundle, hostedServices and plugins explicitly; unknown sections are rejected') for section in [spec['bundle'], *spec['hostedServices'], *spec['plugins'], *([spec['runtime']] if 'runtime' in spec else [])]: for key in ['workspaceId', 'projectId', 'artifactId', 'environmentId']: if key in section: require(type(section[key]) is int and section[key] > 0, 'Owner IDs must be positive integers') report = {'collectedAt': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'verificationScope': 'DECLARED_RELEASE_REFERENCES', 'checks': [], 'businessVerification': 'NOT_EXECUTED', 'notice': 'This report is release evidence, not installation state or business readiness.'} def check(name, action): try: evidence = action() report['checks'].append({'name': name, 'status': 'PASSED', 'evidence': evidence}) return evidence except (ValueError, KeyError, TypeError, OSError, json.JSONDecodeError): report['checks'].append({'name': name, 'status': 'FAILED', 'reason': 'Owner query, identity, contract or expected-state validation failed'}) return None bundle = check('bundle', lambda: bundle_evidence(spec['bundle'], query)) for row in spec['hostedServices']: check(f"hosted-service:{row['projectId']}", lambda row=row: hosted_evidence(row, query)) for row in spec['plugins']: check(f"plugin:{row['pluginId']}", lambda row=row: plugin_evidence(row, query)) if 'runtime' in spec: check('runtime-owner', lambda: runtime_evidence(spec['runtime'], bundle, query)) report['referenceVerification'] = ('PASSED' if all(row['status'] == 'PASSED' for row in report['checks']) else 'FAILED') return report def main(): parser = argparse.ArgumentParser(description=__doc__, epilog=( 'Input: bundle={workspaceId,bundleId,version,projectDirectory}; ' 'hostedServices=[{workspaceId,projectId,version,artifactId,environmentId}]; ' 'plugins=[{workspaceId,pluginId,lifecycleProtocol,referenceProtocol,credentialsProtocol,bindingReadProtocol,inventoryProtocol,propertyWriteProtocol}]; ' 'optional runtime={workspaceId,applicationRuntimeId,environmentKey}. ' 'IDs must come from Owner records. Exit 0 verifies declared references only; ' 'business tests remain NOT_EXECUTED. Exit 1 means collection or validation failed.')) parser.add_argument('--spec', required=True, type=Path) parser.add_argument('--output', required=True, type=Path) args = parser.parse_args() report = collect(json.loads(args.spec.read_text())) args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n') raise SystemExit(0 if report['referenceVerification'] == 'PASSED' else 1) if __name__ == '__main__': main()