70-deployment/deploy-engineering-platform.py

#!/usr/bin/env python3
"""Build, stage, restart and verify the Engineering Platform test service."""
from __future__ import annotations
import argparse, hashlib, json, os, shutil, subprocess, sys, tempfile, time
from datetime import datetime, timezone
from pathlib import Path
from urllib.request import urlopen
import re

VALIDATION_DIR = Path(__file__).resolve().parents[1] / '40-validation'
if str(VALIDATION_DIR) not in sys.path:
    sys.path.insert(0, str(VALIDATION_DIR))
from evidence_lifecycle import seal_evidence, seal_failed_evidence, validate_evidence, write_evidence
from evidence_retention import retain_failed

class DeploymentError(RuntimeError): pass
SERVICE_NAME = "engineering-platform-test.service"

def run(cmd, *, cwd=None, env=None, log=None, check=True):
    p=subprocess.run(cmd,cwd=cwd,env=env,text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
    if log is not None:
        log.write('$ '+' '.join(map(str,cmd))+'\n'+p.stdout+'\n'); log.flush()
    if check and p.returncode:
        raise DeploymentError(f"command failed ({p.returncode}): {' '.join(map(str,cmd))}")
    return p


def service_main_pid(log=None):
    """Return the current systemd user-service main PID, or 0 when inactive."""
    p = run(
        ['systemctl', '--user', 'show', SERVICE_NAME, '--property=MainPID', '--value'],
        log=log,
        check=False,
    )
    if p.returncode != 0:
        return 0
    try:
        return int(p.stdout.strip() or '0')
    except ValueError:
        return 0


def listening_pids(host, port, log=None):
    """Return PIDs listening on the configured TCP port using ss evidence."""
    p = run(['ss', '-ltnp'], log=log, check=False)
    if p.returncode != 0:
        raise DeploymentError('cannot inspect listening TCP ports with ss')
    pids = set()
    suffix = f':{port}'
    for line in p.stdout.splitlines():
        fields = line.split()
        if len(fields) < 4 or not fields[3].endswith(suffix):
            continue
        for value in re.findall(r'pid=(\d+)', line):
            pids.add(int(value))
    return pids


def verify_port_ownership(host, port, evidence, log=None):
    """Reject a target port owned by any process other than the managed service."""
    managed_pid = service_main_pid(log=log)
    pids = listening_pids(host, port, log=log)
    foreign = sorted(pid for pid in pids if pid != managed_pid)
    record = {
        'host': host,
        'port': int(port),
        'managedService': SERVICE_NAME,
        'managedPid': managed_pid,
        'listeningPids': sorted(pids),
        'foreignPids': foreign,
        'status': 'PASSED' if not foreign else 'FAILED',
    }
    (evidence / 'port-ownership.json').write_text(json.dumps(record, indent=2) + '\n')
    if foreign:
        raise DeploymentError(
            f'target port {host}:{port} is occupied by foreign process(es): ' +
            ', '.join(map(str, foreign))
        )
    return record

def load_env(path):
    data={}
    for raw in path.read_text().splitlines():
        line=raw.strip()
        if not line or line.startswith('#') or '=' not in line: continue
        k,v=line.split('=',1); data[k.strip()]=v.strip().strip('"').strip("'")
    return data

def sha256(path):
    h=hashlib.sha256()
    with path.open('rb') as f:
        for chunk in iter(lambda:f.read(1024*1024),b''): h.update(chunk)
    return h.hexdigest()

def tree_digest(root):
    h=hashlib.sha256()
    for path in sorted(p for p in root.rglob('*') if p.is_file()):
        rel=path.relative_to(root).as_posix().encode(); h.update(rel+b'\0')
        with path.open('rb') as f:
            for chunk in iter(lambda:f.read(1024*1024),b''): h.update(chunk)
    return h.hexdigest()

def fetch(url):
    with urlopen(url,timeout=10) as r: return r.status,r.read()

def service_active(log):
    return run(['systemctl','--user','is-active','--quiet',SERVICE_NAME],log=log,check=False).returncode == 0

def stop_service(log):
    if service_active(log):
        run(['systemctl','--user','stop',SERVICE_NAME],log=log)
    for _ in range(50):
        if not service_active(log): return
        time.sleep(.1)
    raise DeploymentError(f'{SERVICE_NAME} did not stop')

def wait_http(url, expected=None, attempts=50):
    last=None
    for _ in range(attempts):
        try:
            status,body=fetch(url)
            if status==200 and (expected is None or expected in body): return status,body
            last=f'HTTP {status}'
        except Exception as exc: last=str(exc)
        time.sleep(.2)
    raise DeploymentError(f'HTTP verification failed for {url}: {last}')

def atomic_file_install(source, target):
    target.parent.mkdir(parents=True,exist_ok=True)
    staged=target.with_name(target.name+'.new')
    shutil.copy2(source,staged); os.chmod(staged,0o755); os.replace(staged,target)

def atomic_tree_install(source, target):
    target.parent.mkdir(parents=True,exist_ok=True)
    staged=target.parent/(target.name+'.new')
    backup=target.parent/(target.name+'.previous')
    shutil.rmtree(staged,ignore_errors=True); shutil.rmtree(backup,ignore_errors=True)
    shutil.copytree(source,staged)
    if target.exists(): os.replace(target,backup)
    os.replace(staged,target)
    shutil.rmtree(backup,ignore_errors=True)



def snapshot_installation(install: Path, snapshot_root: Path) -> dict:
    """Capture the pre-deployment installation for deterministic rollback."""
    snapshot_root.mkdir(parents=True, exist_ok=True)
    state = {"binaryExisted": False, "webExisted": False}
    binary = install / "bin/engineering-platform"
    web = install / "web"
    if binary.is_file():
        target = snapshot_root / "bin/engineering-platform"
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(binary, target)
        state["binaryExisted"] = True
    if web.is_dir():
        shutil.copytree(web, snapshot_root / "web")
        state["webExisted"] = True
    return state


def rollback_installation(install: Path, snapshot_root: Path, state: dict) -> dict:
    """Restore the captured installation and return structured rollback evidence."""
    outcome = {"attempted": True, "status": "FAILED", "restored": []}
    try:
        binary = install / "bin/engineering-platform"
        web = install / "web"
        if state.get("binaryExisted"):
            atomic_file_install(snapshot_root / "bin/engineering-platform", binary)
            outcome["restored"].append("binary")
        elif binary.exists():
            binary.unlink()
            outcome["restored"].append("binary-absence")
        if state.get("webExisted"):
            atomic_tree_install(snapshot_root / "web", web)
            outcome["restored"].append("web")
        elif web.exists():
            shutil.rmtree(web)
            outcome["restored"].append("web-absence")
        outcome["status"] = "SUCCEEDED"
    except Exception as exc:
        outcome["error"] = str(exc)
    return outcome

def prepare_frontend(frontend, tmp, log):
    package=json.loads((frontend/'package.json').read_text())
    scripts=package.get('scripts',{})
    run(['npm','test'],cwd=frontend,log=log)
    if 'build' in scripts:
        if (frontend/'package-lock.json').exists(): run(['npm','ci'],cwd=frontend,log=log)
        run(['npm','run','build'],cwd=frontend,log=log)
        configured=os.environ.get('EP_FRONTEND_BUILD_DIR','dist')
        built=frontend/configured
        if not built.is_dir(): raise DeploymentError(f'frontend build directory missing: {built}')
        mode='COMPILED'
        source=built
    else:
        source=tmp/'frontend-source'
        shutil.copytree(frontend,source,ignore=shutil.ignore_patterns('node_modules','tests','package-lock.json'))
        mode='SOURCE_MODULES'
    return source,mode

def main():
    ap=argparse.ArgumentParser()
    ap.add_argument('--environment',default='test',choices=['test', 'production'])
    ap.add_argument('--engineering-home',default=os.environ.get('ENGINEERING_HOME',str(Path.home()/'engineering')))
    ap.add_argument('--roundtrip-id',default=os.environ.get('NS_ROUNDTRIP_ID','manual'))
    ap.add_argument('--evidence-copy',type=Path)
    ap.add_argument('--repository-roots',type=Path)
    ap.add_argument('--reuse-evidence',type=Path)
    ap.add_argument('--work-item',default='manual')
    ap.add_argument('--chain',default='RC-0001')
    ap.add_argument('--retention-index',type=Path)
    ap.add_argument('--retained-root',type=Path)
    args=ap.parse_args()
    global SERVICE_NAME
    SERVICE_NAME = f"engineering-platform-{args.environment}.service"
    home=Path(args.engineering_home).resolve(); source=home/'engineering-platform'; runtime=home/'runtime'/args.environment
    roots_container=(args.repository_roots or home).resolve()
    if args.reuse_evidence:
        freshness_errors=validate_evidence(roots_container,args.reuse_evidence)
        if freshness_errors:
            raise DeploymentError('reuse evidence is not fresh: '+'; '.join(freshness_errors))
    env_file=runtime/'config/engineering-platform.env'
    if not env_file.exists(): raise DeploymentError(f'missing environment file: {env_file}')
    cfg=load_env(env_file); host=cfg.get('EP_HTTP_HOST','127.0.0.1'); port=cfg.get('EP_HTTP_PORT','18082')
    install=runtime/'installations/engineering-platform'; deployment_id=datetime.now(timezone.utc).strftime('DEP-%Y%m%d-%H%M%S')
    evidence=runtime/'reports/deployment'/deployment_id; evidence.mkdir(parents=True,exist_ok=True)
    statuses={}; log_path=evidence/'deployment.log'; service_was_active=False
    snapshot_root=evidence/'rollback-snapshot'; snapshot_state={}; install_changed=False; failure_phase='initialization'
    try:
      with log_path.open('w') as log:
        failure_phase='validation'; backend=source/'10-backend'; frontend=source/'20-frontend'
        run(['go','test','./...'],cwd=backend,log=log); statuses['goTest']='PASSED'
        run(['go','vet','./...'],cwd=backend,log=log); statuses['goVet']='PASSED'
        tmp=Path(tempfile.mkdtemp(prefix='ep-deploy-'))
        try:
          binary=tmp/'engineering-platform'; run(['go','build','-trimpath','-o',str(binary),'./cmd/engineering-platform'],cwd=backend,log=log)
          digest=sha256(binary); statuses['backendBuild']='PASSED'
          web_source,frontend_mode=prepare_frontend(frontend,tmp,log); web_digest=tree_digest(web_source)
          statuses['javascriptTest']='PASSED'; statuses['frontendBuild']=frontend_mode
          if args.environment == 'production':
            published_reports=install/'web'/'reports'
            if not published_reports.is_dir():
              raise DeploymentError('production documentation reports missing; run the local promotion entrypoint')
            shutil.copytree(published_reports,web_source/'reports')
            statuses['documentationReports']='STAGED'
          unit_template=(home/f'engineering-tools/70-deployment/systemd/engineering-platform-{args.environment}.service.template').read_text()
          unit=unit_template.replace('@ENV_FILE@',str(env_file)).replace('@INSTALL_ROOT@',str(install)).replace('@RUNTIME_ROOT@',str(runtime))
          user_unit=Path.home()/'.config/systemd/user'/SERVICE_NAME; user_unit.parent.mkdir(parents=True,exist_ok=True)
          user_unit.write_text(unit); (evidence/SERVICE_NAME).write_text(unit)
          run(['systemctl','--user','daemon-reload'],log=log)
          verify_port_ownership(host,port,evidence,log=log); statuses['portOwnership']='PASSED'
          failure_phase='installation'
          snapshot_state=snapshot_installation(install,snapshot_root)
          service_was_active=service_active(log); stop_service(log); statuses['serviceStop']='PASSED'
          atomic_file_install(binary,install/'bin/engineering-platform'); install_changed=True
          marker={'deploymentId':deployment_id,'roundtripId':args.roundtrip_id,'frontendSha256':web_digest}
          (web_source/'deployment-marker.json').write_text(json.dumps(marker,indent=2)+'\n')
          atomic_tree_install(web_source,install/'web'); statuses['install']='PASSED'
          run(['systemctl','--user','enable',SERVICE_NAME],log=log)
          failure_phase='startup-verification'
          run(['systemctl','--user','start',SERVICE_NAME],log=log); statuses['serviceStart']='PASSED'
          run(['systemctl','--user','is-active','--quiet',SERVICE_NAME],log=log); statuses['service']='PASSED'
          health_url=f'http://{host}:{port}/health'; status,body=wait_http(health_url)
          (evidence/'health.json').write_bytes(body); parsed=json.loads(body)
          if parsed.get('status')!='UP': raise DeploymentError('health verification failed')
          statuses['health']='PASSED'
          status,body=wait_http(f'http://{host}:{port}/',b'Engineering')
          (evidence/'frontend-smoke.html').write_bytes(body[:65536]); statuses['smoke']='PASSED'
          _,marker_body=wait_http(f'http://{host}:{port}/deployment-marker.json',deployment_id.encode())
          deployed_marker=json.loads(marker_body)
          if deployed_marker.get('frontendSha256') != web_digest: raise DeploymentError('deployed frontend digest mismatch')
          statuses['frontendDeployment']='PASSED'
          if args.environment == 'production':
            _,reports_body=wait_http(f'http://{host}:{port}/reports/', b'Netzwerksolution')
            (evidence/'reports-smoke.html').write_bytes(reports_body[:65536]); statuses['documentationReports']='PASSED'
          journal=run(['journalctl','--user','-u',SERVICE_NAME,'-n','100','--no-pager'],log=log).stdout
          (evidence/'journal.log').write_text(journal)
          metadata={'deploymentId':deployment_id,'roundtripId':args.roundtrip_id,'environment':args.environment,'service':SERVICE_NAME,'healthUrl':health_url,'binarySha256':digest,'frontendSha256':web_digest,'frontendMode':frontend_mode,'status':'SUCCESS','phases':statuses}
          write_evidence(install/'deployment.json',metadata)
          sealed=seal_evidence(
              metadata, roots_container,
              'engineering-tools:70-deployment/deploy-engineering-platform.py',
              [f'runtime:{args.environment}/installations/engineering-platform/deployment.json'],
              work_item=args.work_item, chain=args.chain,
          )
          write_evidence(evidence/'deployment.json',sealed)
        finally: shutil.rmtree(tmp,ignore_errors=True)
    except Exception as exc:
      rollback={"attempted":False,"status":"NOT_REQUIRED"}
      if install_changed:
        rollback=rollback_installation(install,snapshot_root,snapshot_state)
        if rollback.get("status")=="SUCCEEDED" and service_was_active:
          try:
            subprocess.run(['systemctl','--user','start',SERVICE_NAME],check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
            rollback["serviceRestore"]="SUCCEEDED"
          except Exception as service_exc:
            rollback["serviceRestore"]="FAILED"; rollback["serviceError"]=str(service_exc)
      metadata={'evidenceId':'engineering-platform-deployment','deploymentId':deployment_id,'roundtripId':args.roundtrip_id,'environment':args.environment,'status':'FAILED','error':str(exc),'phases':statuses,'createdAt':datetime.now(timezone.utc).isoformat()}
      sealed=seal_failed_evidence(metadata,roots_container,'engineering-tools:70-deployment/deploy-engineering-platform.py',[],work_item=args.work_item,chain=args.chain,failure_phase=failure_phase,rollback=rollback)
      write_evidence(evidence/'deployment.json',sealed)
      retained_root=args.retained_root or runtime/'reports/retained'
      retention_index=args.retention_index or runtime/'reports/evidence-retention-index.json'
      retain_failed(evidence/'deployment.json',retained_root,retention_index)
      if args.evidence_copy:
        args.evidence_copy.mkdir(parents=True,exist_ok=True); shutil.copytree(evidence,args.evidence_copy/deployment_id,dirs_exist_ok=True)
      print(f'Deployment evidence: {evidence}',file=sys.stderr); raise
    shutil.rmtree(snapshot_root,ignore_errors=True)
    report=['# Engineering Platform Deployment Evidence','',f'- Deployment: `{deployment_id}`',f'- Roundtrip: `{args.roundtrip_id}`','- Result: `SUCCESS`','', '## Phases','']+[f'- {k}: `{v}`' for k,v in statuses.items()]
    (evidence/'deployment-report.md').write_text('\n'.join(report)+'\n')
    if args.evidence_copy:
      args.evidence_copy.mkdir(parents=True,exist_ok=True); shutil.copytree(evidence,args.evidence_copy/deployment_id,dirs_exist_ok=True)
    print(f'Deployment evidence: {evidence}'); return 0

if __name__=='__main__':
    try: raise SystemExit(main())
    except DeploymentError as e: print(f'ERROR: {e}',file=sys.stderr); raise SystemExit(1)