"""Owner-configured entrypoint for the existing dedicated Temporal release worker. No caller-supplied transport factory, signing key, command or admission endpoint. A trusted deployment file is configuration, not proof of native authority/custody. The actual principal, key issuers and isolated denial/rollback acceptance remain owner admission requirements before this process can be deployed. """ from __future__ import annotations import argparse import asyncio import base64 import json import os from pathlib import Path import sqlite3 import stat from temporalio.client import Client from temporalio.exceptions import ApplicationError from .release_broker import Broker, Receipts, ROLES, canonical from .release_dispatch import ReleaseActivities, release_worker from .release_sampler import CLUSTER_UID, Probe from .release_transport import GitArgoBackend TEMPORAL_HOST = 'actcore-temporal.activity-core.svc.cluster.local:7233' FIELDS = {'schema_version','enabled','principal','cluster_uid','state_directory', 'kubeconfig','git_config','receipt_trust'} def trusted_file(path): path = Path(path) status = path.lstat() if (not stat.S_ISREG(status.st_mode) or status.st_uid != os.geteuid() or stat.S_IMODE(status.st_mode) & 0o077): raise ValueError('trusted deployment file must be owned and private') return path class Admission: def __init__(self, path): self.path=trusted_file(path) self.value=json.loads(self.path.read_text()) value=self.value if (set(value)!=FIELDS or value['schema_version']!='activity-core-release-runtime/v1' or value['enabled'] is not True or value['principal']!='activity-core-release-broker' or value['cluster_uid']!=CLUSTER_UID): raise ValueError('runtime configuration is disabled or outside admitted scope') self.kubeconfig=trusted_file(value['kubeconfig']) self.git_config=trusted_file(value['git_config']) self.configuration=canonical({k:v for k,v in value.items() if k!='enabled'}) # Retain only digests for drift/revocation checks, never credential copies. import hashlib self.custody={str(path):hashlib.sha256(path.read_bytes()).hexdigest() for path in (self.kubeconfig,self.git_config)} self.root=Path(value['state_directory']) self.root.mkdir(parents=True,exist_ok=True,mode=0o700) status=self.root.lstat() if not stat.S_ISDIR(status.st_mode) or status.st_uid!=os.geteuid() or stat.S_IMODE(status.st_mode)&0o077: raise ValueError('runtime state directory must be owned and private') self.trust={} for key_id,entry in value['receipt_trust'].items(): if (set(entry)!={'public_key','principal','roles'} or not entry['principal'] or not entry['roles'] or not set(entry['roles'])<=ROLES): raise ValueError('invalid deployment-owned receipt trust') public=base64.b64decode(entry['public_key'],validate=True) if len(public)!=32:raise ValueError('invalid Ed25519 public key') self.trust[key_id]={'public_key':public,'principal':entry['principal'],'roles':set(entry['roles'])} if set().union(*(entry['roles'] for entry in self.trust.values()))!=ROLES: raise ValueError('all independent receipt roles require trusted issuers') def require(self): import hashlib current=json.loads(trusted_file(self.path).read_text()) if (current.get('enabled') is not True or canonical({k:v for k,v in current.items() if k!='enabled'})!=self.configuration): raise ValueError('runtime admission disabled or deployment configuration changed') if any(hashlib.sha256(trusted_file(path).read_bytes()).hexdigest()!=digest for path,digest in self.custody.items()): raise ValueError('credential custody files changed; owner re-admission required') def environment(self): # Prevent fallback to an unrelated global Git credential helper/kubeconfig. return {'KUBECONFIG':str(self.kubeconfig),'GIT_CONFIG_GLOBAL':str(self.git_config), 'GIT_CONFIG_NOSYSTEM':'1','GIT_TERMINAL_PROMPT':'0'} class DurableAudit: """Private durable idempotent local receipt sink; no raw transport payloads.""" KEYS={'event_id','event_type','release_id','application','phase','observed_at','candidate','rollback'} def __init__(self,path): self.path=Path(path) with self.connect() as db: db.execute('CREATE TABLE IF NOT EXISTS events (event_id TEXT PRIMARY KEY, payload TEXT NOT NULL)') def connect(self): db=sqlite3.connect(self.path,timeout=5) db.execute('PRAGMA synchronous=FULL') return db def __call__(self,event): if set(event)!=self.KEYS or event['application']!='activity-core' or event['event_type']!='release_transition': raise ValueError('audit event outside release scope') payload=canonical(event).decode() with self.connect() as db: prior=db.execute('SELECT payload FROM events WHERE event_id=?',(event['event_id'],)).fetchone() if prior and prior[0]!=payload:raise ValueError('audit event identity conflict') db.execute('INSERT OR IGNORE INTO events VALUES (?,?)',(event['event_id'],payload)) def verify_runtime_identity(runner): """Check the real configured Kubernetes subject and exact allow/deny boundary.""" namespace=json.loads(runner(['kubectl','--request-timeout=20s','get','namespace','kube-system','-o','json'],timeout=25)) if namespace.get('metadata',{}).get('uid')!=CLUSTER_UID: raise ValueError('runtime cluster UID unverified') identity=json.loads(runner(['kubectl','--request-timeout=20s','auth','whoami','-o','json'],timeout=25)) user=identity.get('status',{}).get('userInfo',{}) if (user.get('username')!='system:serviceaccount:activity-core:activity-core-release-broker' or 'system:masters' in user.get('groups',[])): raise ValueError('actual release principal differs from dedicated service account') allowed=[('get','argoproj.io','applications','argocd','activity-core'), ('patch','argoproj.io','applications','argocd','activity-core'), ('get','argoproj.io','applications','argocd','railiance-apps-root'), ('patch','argoproj.io','applications','argocd','railiance-apps-root')] denied=[('get','','secrets','activity-core',''),('get','','secrets','argocd',''), ('patch','argoproj.io','applications','argocd','target-revenue'), ('patch','apps','deployments','activity-core','actcore-worker'), ('create','','pods/exec','activity-core',''), ('create','rbac.authorization.k8s.io','rolebindings','activity-core',''), ('create','batch','jobs','activity-core','')] for expected,checks in [(True,allowed),(False,denied)]: for verb,group,resource,namespace,name in checks: attrs={'verb':verb,'group':group,'resource':resource,'namespace':namespace} if '/' in resource:attrs['resource'],attrs['subresource']=resource.split('/',1) if name:attrs['name']=name body={'apiVersion':'authorization.k8s.io/v1','kind':'SelfSubjectAccessReview', 'spec':{'resourceAttributes':attrs}} result=json.loads(runner(['kubectl','--request-timeout=20s','create', '--raw=/apis/authorization.k8s.io/v1/selfsubjectaccessreviews','-f','-'], input=json.dumps(body),timeout=25)) status=result.get('status',{}) if status.get('allowed') is not expected or status.get('evaluationError'): raise ValueError('release principal allow/deny boundary unverified') return {'principal_verified':True,'allowed_checks':len(allowed),'denied_checks':len(denied)} class RuntimeActivities(ReleaseActivities): def __init__(self,admission): self.admission=admission broker=Broker(admission.root/'release.sqlite',Receipts(admission.trust),admitted=True) env=admission.environment() from .release_transport import run def bounded_run(argv,**kwargs): admission.require() inherited=kwargs.get('env',os.environ) clean={key:value for key,value in inherited.items() if not key.startswith(('GIT_','KUBE')) and key!='SSH_AUTH_SOCK'} kwargs['env']={**clean,**env} return run(argv,**kwargs) self.identity_probe=lambda:verify_runtime_identity(bounded_run) self.identity_probe() self.probe=Probe(runner=bounded_run) def factory(plan): admission.require() self.identity_probe() def invariants(target): observed,healthy=self.probe() return observed==target and healthy is True backend=GitArgoBackend(plan,admission.root/'transport',invariants, cluster_uid=CLUSTER_UID,admitted=True,runner=bounded_run) return backend super().__init__(broker,factory,DurableAudit(admission.root/'audit.sqlite')) def _plan(self,release_id): try:self.admission.require() except Exception: self.broker.admitted=False raise ApplicationError('release runtime admission unavailable',non_retryable=True) from None return super()._plan(release_id) async def serve(path): admission=Admission(path) activities=RuntimeActivities(admission) client=await asyncio.wait_for(Client.connect(TEMPORAL_HOST,namespace='default'),timeout=30) await release_worker(client,activities).run() def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--configuration',type=Path,required=True) args=parser.parse_args() try:asyncio.run(serve(args.configuration)) except Exception: print(json.dumps({'schema_version':'activity-core-release-runtime/v1','status':'unverified', 'error':'runtime unavailable; private configuration and transport details suppressed'})) return 1 return 0 if __name__=='__main__':raise SystemExit(main())