truf-server/app/sync_alive_github_tokens.py

import sys

sys.dont_write_bytecode = True

import argparse
import os
import re
import threading

import yaml

from migrate_runtime_safety import require_runtime_hardening_stopped
from db_backend import database_url_from_env, is_postgres_url
from paths import apply_path_config
from postgres_runtime import load_postgres_environment
from runtime_security import (
    ClusterAuthorityLock,
    canonical_path,
    durable_replace,
    harden_private_file,
    PrivateFileLock,
    private_file_ready,
    require_private_directory,
    require_private_file,
)


GITHUB_TOKEN_PREFIXES = ('ghp_', 'gho_', 'ghu_', 'ghs_', 'ghr_', 'github_pat_')
ACCEPTED_ALIVE_STATUSES = {'ALIVE', 'VALID', 'VALID_2FA'}
DOCKERHUB_TOKEN_RE = re.compile(r'^dckr_pat_[A-Za-z0-9_-]{27}$')
PROVIDER_DEFAULTS = {
    'github': {
        'alive_file': os.path.join('runtime', 'keychecks', 'github', 'githubAlive.txt'),
        'pool': 'github_main',
        'name_prefix': 'gh',
    },
    'dockerhub': {
        'alive_file': os.path.join('runtime', 'keychecks', 'dockerhub', 'dockerhubAlive.txt'),
        'pool': 'dockerhub_main',
        'name_prefix': 'dockerhub',
    },
}


def read_alive_tokens(path):
    tokens = []
    seen = set()
    with open(path, 'r', encoding='utf-8', errors='replace') as f:
        for line in f:
            # Status files are TSV-like: token, status, message, extra.
            fields = line.rstrip('\r\n').split('\t')
            token = fields[0].strip() if fields else ''
            if not token or not token.startswith(GITHUB_TOKEN_PREFIXES):
                continue
            if any(character.isspace() for character in token):
                raise ValueError('alive token input contains whitespace in a token field')
            status = fields[1].strip().upper() if len(fields) > 1 else ''
            if status not in ACCEPTED_ALIVE_STATUSES:
                raise ValueError(f'alive token input contains an unaccepted or missing status: {status or "(missing)"}')
            if token in seen:
                continue
            seen.add(token)
            tokens.append(token)
    return tokens


def read_alive_credentials(path, provider):
    provider = str(provider or 'github').strip().lower()
    if provider == 'github':
        return [{'token': token} for token in read_alive_tokens(path)], 0
    if provider != 'dockerhub':
        raise ValueError(f'unsupported alive credential provider: {provider}')

    credentials = []
    seen = {}
    skipped_missing_username = 0
    with open(path, 'r', encoding='utf-8', errors='replace') as handle:
        for line in handle:
            fields = line.rstrip('\r\n').split('\t')
            identity = fields[0].strip() if fields else ''
            if not identity:
                continue
            if ':' in identity:
                username, token = identity.rsplit(':', 1)
                username = username.strip()
                token = token.strip()
            else:
                username = ''
                token = identity
            if not DOCKERHUB_TOKEN_RE.fullmatch(token):
                continue
            status = fields[1].strip().upper() if len(fields) > 1 else ''
            if status not in {'VALID', 'VALID_2FA'}:
                raise ValueError(f'alive DockerHub input contains an unaccepted or missing status: {status or "(missing)"}')
            if not username:
                skipped_missing_username += 1
                continue
            if len(username) > 256 or ':' in username or any(character.isspace() for character in username):
                raise ValueError('alive DockerHub input contains an invalid username field')
            previous = seen.get(token)
            if previous is not None:
                if previous.casefold() != username.casefold():
                    raise ValueError('alive DockerHub input contains conflicting usernames for one token')
                continue
            seen[token] = username
            credentials.append({'username': username, 'token': token})
    return credentials, skipped_missing_username


def next_name(existing_names, prefix):
    pattern = re.compile(rf'^{re.escape(prefix)}_(\d+)$')
    max_index = 0
    for name in existing_names:
        match = pattern.match(str(name or ''))
        if match:
            max_index = max(max_index, int(match.group(1)))
    return f'{prefix}_{max_index + 1}'


def _atomic_write_private_yaml(path, value):
    parent = require_private_directory(os.path.dirname(os.path.abspath(path)), create=False)
    payload = yaml.safe_dump(value, allow_unicode=True, sort_keys=False, width=120).encode('utf-8')
    temporary = f'{path}.{os.getpid()}.{threading.get_ident()}.tmp'
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, 'O_BINARY', 0) | getattr(os, 'O_NOFOLLOW', 0)
    descriptor = os.open(temporary, flags, 0o600)
    try:
        os.close(descriptor)
        descriptor = None
        harden_private_file(temporary)
        with open(temporary, 'wb') as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        if not private_file_ready(temporary):
            raise OSError(f'private temporary secrets ACL changed: {temporary}')
        durable_replace(temporary, path)
        if not private_file_ready(path):
            raise OSError(f'private secrets ACL changed during publication: {path}')
    finally:
        if descriptor is not None:
            os.close(descriptor)
        try:
            if os.path.exists(temporary):
                os.remove(temporary)
        except OSError:
            pass


def sync_tokens(
    secrets_path,
    alive_path,
    pool_name,
    name_prefix,
    apply=False,
    *,
    provider='github',
    replace_conflicting_usernames=False,
    canonical_secrets_path=None,
    authority_lock=None,
    stopped_verified=False,
):
    if authority_lock is None or not getattr(authority_lock, 'acquired', False):
        raise RuntimeError('alive-token sync requires an acquired cluster authority lock')
    if stopped_verified is not True:
        raise RuntimeError('alive-token sync requires verified stopped runtime proof')
    if not canonical_secrets_path or canonical_path(secrets_path) != canonical_path(canonical_secrets_path):
        raise RuntimeError('alive-token sync secrets path must exactly match canonical global.secrets_file')
    if not os.path.exists(alive_path):
        raise SystemExit(f'alive token file not found: {alive_path}')
    if not os.path.exists(secrets_path):
        raise SystemExit(f'secrets file not found: {secrets_path}')

    require_private_file(secrets_path)
    require_private_file(alive_path)
    lock = PrivateFileLock(f'{secrets_path}.sync.lock').acquire()
    try:
        require_private_file(secrets_path)
        require_private_file(alive_path)
        with open(secrets_path, 'r', encoding='utf-8') as f:
            secrets = yaml.safe_load(f) or {}

        auth_pools = secrets.setdefault('auth_pools', {})
        pool = auth_pools.setdefault(pool_name, [])
        if not isinstance(pool, list):
            raise SystemExit(f'auth_pools.{pool_name} must be a list')
        existing_before = len(pool)

        provider = str(provider or 'github').strip().lower()
        if provider not in PROVIDER_DEFAULTS:
            raise ValueError(f'unsupported alive credential provider: {provider}')
        existing_tokens = set()
        existing_entries = {}
        existing_names = set()
        normalized_existing = 0
        normalized_usernames = 0
        for entry in pool:
            if not isinstance(entry, dict):
                continue
            if entry.get('name'):
                existing_names.add(str(entry.get('name')))
            token = str(entry.get('token') or '')
            stripped = token.strip()
            if stripped != token:
                normalized_existing += 1
                if apply:
                    entry['token'] = stripped
            if stripped:
                existing_tokens.add(stripped)
                existing_entries.setdefault(stripped, []).append(entry)
            if provider == 'dockerhub':
                username = str(entry.get('username') or '')
                stripped_username = username.strip()
                if stripped_username != username:
                    normalized_usernames += 1
                    if apply:
                        entry['username'] = stripped_username

        alive_credentials, skipped_missing_username = read_alive_credentials(alive_path, provider)
        added = []
        username_filled = 0
        username_conflicts = 0
        username_replaced = 0
        for credential in alive_credentials:
            token = credential['token']
            if token in existing_tokens:
                if provider == 'dockerhub':
                    entries = existing_entries.get(token, [])
                    usernames = {
                        str(entry.get('username') or '').strip().casefold()
                        for entry in entries if str(entry.get('username') or '').strip()
                    }
                    expected = credential['username'].casefold()
                    if len(usernames) > 1 or (usernames and expected not in usernames):
                        if replace_conflicting_usernames:
                            username_replaced += 1
                            if apply:
                                for entry in entries:
                                    entry['username'] = credential['username']
                        else:
                            username_conflicts += 1
                        continue
                    if not usernames:
                        username_filled += 1
                        if apply:
                            for entry in entries:
                                entry['username'] = credential['username']
                continue
            name = next_name(existing_names, name_prefix)
            existing_names.add(name)
            existing_tokens.add(token)
            new_entry = {'name': name}
            if provider == 'dockerhub':
                new_entry['username'] = credential['username']
            new_entry['token'] = token
            added.append(new_entry)

        if apply and (
            added or normalized_existing or normalized_usernames
            or username_filled or username_replaced
        ):
            pool.extend(added)
            _atomic_write_private_yaml(secrets_path, secrets)

        return {
            'alive_unique': len(alive_credentials),
            'existing_before': existing_before,
            'added': len(added),
            'normalized_existing': normalized_existing,
            'normalized_usernames': normalized_usernames,
            'username_filled': username_filled,
            'username_conflicts': username_conflicts,
            'username_replaced': username_replaced,
            'skipped_missing_username': skipped_missing_username,
            'pool_after': len(pool) + (0 if apply else len(added)),
        }
    finally:
        lock.release()


def load_config(path):
    with open(path, 'r', encoding='utf-8') as handle:
        return apply_path_config(yaml.safe_load(handle) or {}, path)


def parse_args():
    parser = argparse.ArgumentParser(description='Sync alive provider credentials into a private auth pool without printing them.')
    parser.add_argument('--provider', choices=sorted(PROVIDER_DEFAULTS), default='github')
    parser.add_argument('--secrets', help='Must exactly match global.secrets_file from --config')
    parser.add_argument('--alive-file')
    parser.add_argument('--pool')
    parser.add_argument('--name-prefix')
    parser.add_argument('--config', default=os.path.join('app', 'config.yaml'))
    parser.add_argument('--dry-run', action='store_true')
    parser.add_argument('--apply', action='store_true', help='Apply under verified offline maintenance authority')
    parser.add_argument(
        '--replace-conflicting-usernames', action='store_true',
        help='Replace an existing DockerHub username only when the same token has an authoritative alive pair',
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.apply and args.dry_run:
        raise SystemExit('--apply and --dry-run are mutually exclusive')
    config_path = canonical_path(args.config)
    require_private_file(config_path)
    config = load_config(config_path)
    configured_value = (config.get('global') or {}).get('secrets_file')
    if not configured_value:
        raise SystemExit('global.secrets_file is required')
    configured_secrets = canonical_path(configured_value)
    requested_secrets = canonical_path(args.secrets) if args.secrets else configured_secrets
    if requested_secrets != configured_secrets:
        raise SystemExit('--secrets must exactly match canonical global.secrets_file')
    load_postgres_environment(config_path, config)
    endpoint_dsn = database_url_from_env() or (config.get('global') or {}).get('database_url')
    if not is_postgres_url(endpoint_dsn):
        raise SystemExit('A caller-selected canonical PostgreSQL DSN is required for maintenance authority')
    provider = str(getattr(args, 'provider', 'github') or 'github').strip().lower()
    defaults = PROVIDER_DEFAULTS.get(provider)
    if defaults is None:
        raise SystemExit(f'unsupported provider: {provider}')
    alive_file = args.alive_file or defaults['alive_file']
    pool_name = args.pool or defaults['pool']
    name_prefix = args.name_prefix or defaults['name_prefix']
    with ClusterAuthorityLock(config, endpoint_dsn=endpoint_dsn) as authority_lock:
        require_runtime_hardening_stopped(config)
        result = sync_tokens(
            configured_secrets,
            alive_file,
            pool_name,
            name_prefix,
            apply=args.apply,
            provider=provider,
            replace_conflicting_usernames=bool(getattr(args, 'replace_conflicting_usernames', False)),
            canonical_secrets_path=configured_secrets,
            authority_lock=authority_lock,
            stopped_verified=True,
        )
    mode = 'updated' if args.apply else 'dry_run'
    print(
        f"{mode}: provider={provider} pool={pool_name} alive_unique={result['alive_unique']} "
        f"existing_before={result['existing_before']} added={result['added']} "
        f"username_filled={result.get('username_filled', 0)} "
        f"username_conflicts={result.get('username_conflicts', 0)} "
        f"username_replaced={result.get('username_replaced', 0)} "
        f"skipped_missing_username={result.get('skipped_missing_username', 0)} "
        f"normalized_existing={result['normalized_existing']} "
        f"normalized_usernames={result.get('normalized_usernames', 0)} pool_after={result['pool_after']}"
    )
    return 0


if __name__ == '__main__':
    main()