kangshifu-board/deploy/enterprise-workshop/registry.py

"""Host-side enterprise -> pinned Gogs checkout mapping and post-initialization entry."""
import argparse
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import uuid

sys.dont_write_bytecode = True
from bootstrap import COMPANY, Stopped, atomic_create, encode, need, plain, private_json, sealed, sha
from host_bootstrap import initialize, run


def register(registry, checkout, commit, apply=False, runner=subprocess.run):
    registry, checkout = plain(registry), plain(checkout)
    cfg, _ = sealed(checkout)
    need(re.fullmatch('[a-f0-9]{40}', commit), 'PINNED_COMMIT_REQUIRED')
    need(run(['git', '-C', str(checkout), 'rev-parse', 'HEAD'], runner).decode().strip() == commit,
         'CONFIGURATION_COMMIT_MISMATCH')
    remote = run(['git', '-C', str(checkout), 'remote', 'get-url', 'origin'], runner).decode().strip()
    expected = 'https://git.fmode.cn/' + cfg['organization'] + '/' + cfg['repository'] + '.git'
    need(remote == expected, 'CONFIGURATION_REPOSITORY_MISMATCH')
    need(not run(['git', '-C', str(checkout), 'status', '--porcelain'], runner).strip(),
         'CONFIGURATION_CHECKOUT_CHANGED')
    before = registry.read_bytes() if registry.exists() else None
    mapping = private_json(registry) if before is not None else {'schema': 'ksf.enterprise-registry.v1', 'companies': {}}
    need(set(mapping) == {'schema', 'companies'} and mapping['schema'] == 'ksf.enterprise-registry.v1'
         and isinstance(mapping['companies'], dict), 'INVALID_ENTERPRISE_REGISTRY')
    need(checkout != registry and checkout not in registry.parents, 'REGISTRY_INSIDE_CHECKOUT')
    entry = {'repositoryURL': expected, 'checkout': str(checkout), 'commit': commit,
             'packageManifestSha256': sha((checkout / 'FILES.sha256').read_bytes())}
    changed = mapping['companies'].get(COMPANY) != entry
    mapping['companies'][COMPANY] = entry
    if changed and apply:
        if before is not None:
            atomic_create(registry.with_name(registry.name + '.before-' + uuid.uuid4().hex), before)
        temp = registry.with_name('.' + registry.name + '.' + uuid.uuid4().hex)
        try:
            atomic_create(temp, encode(mapping))
            need((registry.read_bytes() if registry.exists() else None) == before, 'REGISTRY_CONCURRENT_CHANGE')
            os.replace(temp, registry)
        finally:
            if temp.exists():
                temp.unlink()
    return {'ok': True, 'planOnly': not apply, 'mappingChanged': changed and apply,
            'companyObjectId': COMPANY, 'configuration': entry, 'businessWrites': 0}


def after_twin(registry, container, apply=False, runner=subprocess.run):
    need(re.fullmatch('[a-zA-Z0-9][a-zA-Z0-9_.-]{0,100}', container), 'INVALID_CONTAINER_NAME')
    current = json.loads(run(['docker', 'exec', '--user', '10000:10000', container,
                              'cat', '/opt/data/.fmode/config/enterprise/current.json'], runner))
    mapping = private_json(registry)
    need(mapping.get('schema') == 'ksf.enterprise-registry.v1', 'INVALID_ENTERPRISE_REGISTRY')
    entry = mapping.get('companies', {}).get(current.get('companyObjectId'))
    if entry is None:
        return {'ok': True, 'skipped': True, 'reason': 'NO_ENTERPRISE_WORKSHOP_MAPPING', 'businessWrites': 0}
    need(current.get('companyObjectId') == COMPANY, 'UNSUPPORTED_ENTERPRISE')
    root = plain(entry['checkout'])
    need(sha((root / 'FILES.sha256').read_bytes()) == entry['packageManifestSha256'], 'REGISTERED_PACKAGE_CHANGED')
    return initialize(root, entry['commit'], container, apply, runner)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest='command', required=True)
    bind = sub.add_parser('bind')
    bind.add_argument('--checkout', type=Path, required=True)
    bind.add_argument('--commit', required=True)
    hook = sub.add_parser('after-twin')
    hook.add_argument('--container', required=True)
    for p in (bind, hook):
        p.add_argument('--registry', type=Path, required=True)
        p.add_argument('--apply', action='store_true')
    args = parser.parse_args()
    need(args.registry.is_absolute(), 'ABSOLUTE_PATH_REQUIRED')
    if args.command == 'bind':
        need(args.checkout.is_absolute(), 'ABSOLUTE_PATH_REQUIRED')
        result = register(args.registry, args.checkout, args.commit, args.apply)
    else:
        result = after_twin(args.registry, args.container, args.apply)
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    try:
        main()
    except (Stopped, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as error:
        print(json.dumps({'ok': False, 'error': {'code': str(error) if isinstance(error, Stopped)
                                               else 'ENTERPRISE_REGISTRY_STOPPED'}}), file=sys.stderr)
        sys.exit(1)