kangshifu-board/scripts/install-wecom-tool-result-hotfix.py

"""Exact two-file patch. Default is read-only; never restarts the gateway."""
import argparse
import ast
import hashlib
import json
import os
from pathlib import Path
import stat
import tempfile

ROOT = Path(__file__).resolve().parent.parent
TARGET = Path('/opt/hermes/plugins/platforms/wecom')
BEFORE = {
    'workshop_registration.py': 'ecef06cacb4e05cc84ee3256fd6841f0da2bfa4d1cf69f39a2e6f38b05461d3e',
    'identity_bridge.py': 'fa26f0c8918650a3d68869026ce547fa7ed0e50974b60819c954105ba7b16e01',
}


class PatchError(Exception):
    pass


def digest(data):
    return hashlib.sha256(data).hexdigest()


def plain(path):
    if not path.is_absolute() or path.resolve() != path:
        raise PatchError('UNSAFE_PATCH_PATH')
    for p in (path, *path.parents):
        if p.is_symlink():
            raise PatchError('UNSAFE_PATCH_PATH')
    info = path.lstat()
    if not stat.S_ISREG(info.st_mode):
        raise PatchError('NOT_PLAIN_FILE')
    return info


def atomic(path, content, mode, uid, gid):
    fd, name = tempfile.mkstemp(prefix='.' + path.name + '.hotfix-', dir=path.parent)
    try:
        with os.fdopen(fd, 'wb') as f:
            f.write(content)
            f.flush()
            os.fsync(f.fileno())
            os.fchmod(f.fileno(), mode)
            if hasattr(os, 'fchown'):
                os.fchown(f.fileno(), uid, gid)
        os.replace(name, path)
    finally:
        if os.path.exists(name):
            os.unlink(name)


def install(source_root=ROOT, target=TARGET, apply=False, backup_dir=None):
    manifest = json.loads((source_root / 'HOTFIX.json').read_text(encoding='utf-8'))
    if manifest.get('version') != 'ksf.wecom-tool-result-hotfix.v1' or set(manifest.get('files', {})) != set(BEFORE):
        raise PatchError('INVALID_HOTFIX_MANIFEST')
    entries = []
    for name, old_hash in BEFORE.items():
        origin = source_root / 'integrations/wecom-identity' / name
        plain(origin)
        data = origin.read_bytes()
        expected = manifest['files'][name]
        if expected['beforeSha256'] != old_hash or digest(data) != expected['sha256']:
            raise PatchError('PACKAGE_HASH_MISMATCH')
        ast.parse(data.decode('utf-8'))
        dest = target / name
        info = plain(dest)
        prior = dest.read_bytes()
        if digest(prior) not in (old_hash, digest(data)):
            raise PatchError('UNKNOWN_TARGET_VERSION')
        entries.append({'name': name, 'path': dest, 'data': data, 'prior': prior, 'mode': stat.S_IMODE(info.st_mode), 'uid': info.st_uid, 'gid': info.st_gid})
    journal = {'ok': True, 'readOnly': not apply, 'operation': 'tool-result-hotfix-plan' if not apply else 'tool-result-hotfix-installed',
               'gatewayRestarted': False, 'businessWrites': 0, 'identityStateChanged': False,
               'files': [{'name': e['name'], 'beforeSha256': digest(e['prior']), 'afterSha256': digest(e['data']),
                          'action': 'unchanged' if e['prior'] == e['data'] else 'update'} for e in entries]}
    if not apply:
        return journal
    changed = [e for e in entries if e['prior'] != e['data']]
    if not changed:
        return journal
    if backup_dir is None or not backup_dir.is_absolute() or backup_dir.exists() or backup_dir.parent.resolve() != backup_dir.parent or backup_dir.parent == target or backup_dir.parent == source_root:
        raise PatchError('NEW_ABSOLUTE_BACKUP_REQUIRED')
    if target == backup_dir or target in backup_dir.parents or backup_dir in target.parents:
        raise PatchError('BACKUP_TARGET_OVERLAP')
    for parent in backup_dir.parents:
        if parent.is_symlink():
            raise PatchError('UNSAFE_PATCH_PATH')
    backup_dir.mkdir(mode=0o700)
    journal['backup'] = str(backup_dir)
    for e in changed:
        with (backup_dir / e['name']).open('xb') as f:
            f.write(e['prior'])
        os.chmod(backup_dir / e['name'], 0o600)
    with (backup_dir / 'before.json').open('x', encoding='utf-8') as f:
        json.dump([{'name': e['name'], 'sha256': digest(e['prior']), 'uid': e['uid'], 'gid': e['gid'], 'mode': e['mode']} for e in changed], f)
    os.chmod(backup_dir / 'before.json', 0o600)
    done = []
    try:
        for e in changed:
            info = plain(e['path'])
            if e['path'].read_bytes() != e['prior'] or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (e['uid'], e['gid'], e['mode']):
                raise PatchError('TARGET_CHANGED_CONCURRENTLY')
            done.append(e)
            atomic(e['path'], e['data'], e['mode'], e['uid'], e['gid'])
        for e in entries:
            if e['path'].read_bytes() != e['data']:
                raise PatchError('PATCH_READBACK_MISMATCH')
        return journal
    except Exception:
        complete = True
        for e in reversed(done):
            try:
                plain(e['path'])
                current = e['path'].read_bytes()
                if current == e['prior']:
                    continue
                if current != e['data']:
                    complete = False
                    continue
                atomic(e['path'], e['prior'], e['mode'], e['uid'], e['gid'])
            except Exception:
                complete = False
        error = PatchError('HOTFIX_INSTALL_FAILED')
        error.receipt = {**journal, 'ok': False, 'rollbackComplete': complete}
        raise error from None


if __name__ == '__main__':
    try:
        parser = argparse.ArgumentParser()
        parser.add_argument('--apply', action='store_true')
        parser.add_argument('--backup-dir', type=Path)
        args = parser.parse_args()
        if os.name != 'posix' or os.geteuid() != 0:
            raise PatchError('LINUX_ROOT_REQUIRED')
        print(json.dumps(install(apply=args.apply, backup_dir=args.backup_dir), ensure_ascii=False, indent=2))
    except Exception as exc:
        code = str(exc) if isinstance(exc, PatchError) else 'HOTFIX_STOPPED'
        print(json.dumps(getattr(exc, 'receipt', {'ok': False, 'error': {'code': code}, 'gatewayRestarted': False}), ensure_ascii=False))
        raise SystemExit(1) from None