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

"""Create the enterprise organization/config repository as its designated administrator."""
import argparse
import base64
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request

sys.dont_write_bytecode = True
from bootstrap import HERE, Stopped, need, plain, private_json, sealed, sha

ORIGIN = 'https://git.fmode.cn'


class Gogs:
    def __init__(self, token):
        self.token = token

    def call(self, method, route, body=None, missing=False):
        need(route.startswith('/api/v1/') and '..' not in route, 'INVALID_GOGS_ROUTE')
        request = urllib.request.Request(ORIGIN + route, method=method,
                                        headers={'Authorization': 'token ' + self.token,
                                                 'Content-Type': 'application/json'},
                                        data=None if body is None else json.dumps(body).encode())
        # Credentials may only go to this origin; Gogs API calls do not need redirects.
        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, *args, **kwargs):
                return None
        try:
            with urllib.request.build_opener(NoRedirect()).open(request, timeout=30) as response:
                raw = response.read(2 * 1024 * 1024)
                return json.loads(raw) if raw else None
        except urllib.error.HTTPError as error:
            if error.code == 404 and missing:
                return None
            raise Stopped('GOGS_HTTP_' + str(error.code)) from None


def lookup_organization(api, name):
    org = api.call('GET', '/api/v1/orgs/' + name, missing=True)
    if org:
        return org
    # A missing route is not evidence that an organization does not exist.
    mine = api.call('GET', '/api/v1/user/orgs')
    need(isinstance(mine, list), 'GOGS_ORGANIZATION_LIST_INVALID')
    matches = [row for row in mine if (row.get('username') or row.get('login')) == name]
    need(len(matches) <= 1, 'GOGS_ORGANIZATION_LIST_INVALID')
    if matches:
        return matches[0]
    # This also detects a same-name organization outside the administrator's memberships.
    account = api.call('GET', '/api/v1/users/' + name, missing=True)
    need(account is None, 'EXISTING_ORGANIZATION_MEMBERSHIP_UNVERIFIED')
    return None


def inspect(api, cfg):
    who = api.call('GET', '/api/v1/user')
    need((who.get('username') or who.get('login')) == cfg['administrator'], 'GOGS_ADMINISTRATOR_MISMATCH')
    org = lookup_organization(api, cfg['organization'])
    owner_verified = False
    if org:
        teams = api.call('GET', '/api/v1/orgs/' + cfg['organization'] + '/teams', missing=True)
        if teams is not None:
            owners = [t for t in teams if t['name'] == 'Owners']
            need(len(owners) == 1, 'GOGS_OWNER_TEAM_MISSING')
            members = api.call('GET', '/api/v1/teams/' + str(owners[0]['id']) + '/members', missing=True)
            if members is None:
                members = api.call('GET', '/api/v1/admin/teams/' + str(owners[0]['id']) + '/members')
            need(any((m.get('username') or m.get('login')) == cfg['administrator'] for m in members),
                 'EXISTING_ORGANIZATION_OWNER_MISMATCH')
            owner_verified = True
    repo = api.call('GET', '/api/v1/repos/' + cfg['organization'] + '/' + cfg['repository'], missing=True)
    if repo:
        need(repo.get('private') is True and repo.get('owner', {}).get('username',
             repo.get('owner', {}).get('login')) == cfg['organization'], 'EXISTING_REPOSITORY_MISMATCH')
    return {'organizationExists': bool(org), 'repositoryExists': bool(repo),
            'organizationOwnerVerified': owner_verified,
            'ownerVerificationUnavailable': bool(org) and not owner_verified,
            'administrator': cfg['administrator'], 'organization': cfg['organization'],
            'repository': cfg['repository']}


def git(args, cwd, env):
    result = subprocess.run(['git', *args], cwd=cwd, env=env, capture_output=True, timeout=120)
    need(result.returncode == 0, 'CONFIGURATION_GIT_OPERATION_FAILED')
    return result.stdout


def repository_state(existing, root, files):
    expected = set(files) | {'FILES.sha256'}
    if set(existing) == expected and all(existing[n] == (root / n).read_bytes() for n in expected):
        return 'current'
    prior = json.loads((root / 'previous-release.json').read_bytes())['files']
    need(set(existing) == set(prior) and set(prior) <= expected
         and all(sha(existing[n]) == digest for n, digest in prior.items()),
         'EXISTING_REPOSITORY_CONTENT_CHANGED')
    return 'known-r1'


def create_organization(api, cfg):
    org = cfg['organization']
    description = '企业配置来源；companyObjectId=' + cfg['companyObjectId']
    try:
        return api.call('POST', '/api/v1/orgs',
                        {'username': org, 'full_name': cfg['companyName'], 'description': description})
    except Stopped as error:
        if str(error) != 'GOGS_HTTP_404':
            raise
    # Gogs 0.13 exposes organization creation on the admin user route, with form field names.
    # An uncertain response or authorization failure never triggers a second write attempt.
    existing = lookup_organization(api, org)
    if existing is not None:
        return existing
    return api.call('POST', '/api/v1/admin/users/' + cfg['administrator'] + '/orgs',
                    {'UserName': org, 'FullName': cfg['companyName'], 'Description': description})


def setup(api, root, apply=False):
    root = plain(root)
    cfg, files = sealed(root)
    status = inspect(api, cfg)
    if not apply:
        return {'ok': True, 'planOnly': True, **status, 'productionConfigurationChanged': False}
    org, name = cfg['organization'], cfg['repository']
    if not status['organizationExists']:
        create_organization(api, cfg)
    # Revalidate ownership before creating a repository in an existing/new organization.
    status = inspect(api, cfg)
    need(status['organizationExists'] and status['organizationOwnerVerified'], 'GOGS_OWNER_VERIFICATION_UNAVAILABLE')
    if not status['repositoryExists']:
        api.call('POST', '/api/v1/org/' + org + '/repos',
                 {'name': name, 'description': '康师傅工作坊配置与初始化组件；不存放凭据',
                  'private': True, 'auto_init': False})
    repo = api.call('GET', '/api/v1/repos/' + org + '/' + name)
    need(repo.get('private') is True, 'REPOSITORY_NOT_PRIVATE')
    env = dict(os.environ)
    env.update(GIT_TERMINAL_PROMPT='0', GIT_CONFIG_COUNT='2',
               GIT_CONFIG_KEY_0='http.' + ORIGIN + '/.extraheader',
               GIT_CONFIG_VALUE_0='Authorization: Basic ' + base64.b64encode(
                   (cfg['administrator'] + ':' + api.token).encode()).decode(),
               GIT_CONFIG_KEY_1='http.followRedirects', GIT_CONFIG_VALUE_1='false')
    remote = ORIGIN + '/' + org + '/' + name + '.git'
    with tempfile.TemporaryDirectory(prefix='ksf-enterprise-git-') as temp:
        staging = Path(temp)
        git(['init', '--initial-branch=main'], staging, env)
        git(['config', 'core.autocrlf', 'false'], staging, env)
        git(['remote', 'add', 'origin', remote], staging, env)
        refs = git(['ls-remote', 'origin'], staging, env).decode().splitlines()
        had_content = bool(refs)
        heads = [x for x in refs if x.endswith('\trefs/heads/main')]
        commit = None
        if refs:
            need(len(heads) == 1, 'EXISTING_REPOSITORY_BASELINE_UNKNOWN')
            git(['fetch', '--depth=1', 'origin', 'main'], staging, env)
            tracked = git(['ls-tree', '-r', '--name-only', 'FETCH_HEAD'], staging, env).decode().splitlines()
            existing = {n: git(['show', 'FETCH_HEAD:' + n], staging, env) for n in tracked}
            if repository_state(existing, root, files) == 'current':
                commit = git(['rev-parse', 'FETCH_HEAD'], staging, env).decode().strip()
            else:
                git(['checkout', '-B', 'main', 'FETCH_HEAD'], staging, env)
        changed = commit is None
        if changed:
            for relative in [*files, 'FILES.sha256']:
                dest = staging / relative
                dest.parent.mkdir(parents=True, exist_ok=True)
                dest.write_bytes((root / relative).read_bytes())
            git(['config', 'user.name', 'Kangshifu configuration release'], staging, env)
            git(['config', 'user.email', 'ksf-config@fmode.cn'], staging, env)
            git(['add', '--', *sorted(files), 'FILES.sha256'], staging, env)
            git(['commit', '-m', 'Configure enterprise workshop and required VOC runtime'], staging, env)
            commit = git(['rev-parse', 'HEAD'], staging, env).decode().strip()
            git(['push', 'origin', 'HEAD:refs/heads/main'], staging, env)
        refs = git(['ls-remote', 'origin', 'refs/heads/main'], staging, env).decode()
        need(refs.split()[0] == commit, 'GOGS_PUSH_NOT_CONFIRMED')
        # Read every remote file back, including the private configuration manifest.
        git(['fetch', '--depth=1', 'origin', commit], staging, env)
        for relative in [*files, 'FILES.sha256']:
            need(git(['show', 'FETCH_HEAD:' + relative], staging, env) == (root / relative).read_bytes(),
                 'GOGS_READBACK_MISMATCH')
    return {'ok': True, 'planOnly': False, **inspect(api, cfg), 'repositoryURL': remote,
            'commit': commit, 'repositoryContentCreated': changed and not had_content,
            'repositoryContentUpdated': changed and had_content, 'private': True,
            'filesReadBack': len(files) + 1, 'credentialsInRepository': False,
            'enterprisePlatformBindingInstalled': False, 'businessWrites': 0}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--token-file', required=True, type=Path)
    parser.add_argument('--apply', action='store_true')
    args = parser.parse_args()
    # The supplied private JSON contains only an existing Gogs access token, never a password reset.
    token = private_json(args.token_file).get('token')
    need(isinstance(token, str) and len(token) >= 20 and '\n' not in token, 'GOGS_TOKEN_MISSING')
    print(json.dumps(setup(Gogs(token), HERE, args.apply), ensure_ascii=False, indent=2))


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