kangshifu-board/scripts/probe-unified-target.py

"""Compare private configuration, sealed runtime constants and actual read-only identities.

This diagnostic never changes the migration guard, configuration or database.
Only database/role names and comparison results are returned, never connection data.
"""
import argparse
import ast
import hashlib
import json
import os
from pathlib import Path
import sys

sys.dont_write_bytecode = True


def need(ok, code):
    if not ok:
        raise ValueError(code)


def private_configuration(path):
    need(path.is_absolute() and path.is_file() and not path.is_symlink(), 'PRIVATE_CONFIG_REQUIRED')
    if os.name != 'nt':
        need(path.stat().st_mode & 0o077 == 0, 'PRIVATE_CONFIG_PERMISSIONS_REQUIRED')
    value = json.loads(path.read_text('utf-8-sig'))
    need(set(value) == {'schema', 'appId', 'database', 'connection'} and
         value['schema'] == 'ksf.unified-db.v1' and value['appId'] == 'FKieRbIoGt_main', 'CONFIG_APPLICATION_MISMATCH')
    need(isinstance(value['database'], str) and value['database'], 'CONFIG_DATABASE_REQUIRED')
    allowed = {'service', 'host', 'port', 'user', 'passfile', 'sslmode', 'sslrootcert', 'connect_timeout'}
    connection = value['connection']
    need(isinstance(connection, dict) and connection.get('service') and not set(connection) - allowed,
         'EXISTING_SERVICE_CONFIGURATION_REQUIRED')
    return value


def runtime_identity(root):
    need(root.is_absolute(), 'ABSOLUTE_RUNTIME_ROOT_REQUIRED')
    file = root / 'server/voc-library/unified_store.py'
    raw = file.read_bytes()
    tree = ast.parse(raw.decode('utf-8-sig'))
    values = {}
    for statement in tree.body:
        if isinstance(statement, ast.Assign):
            for target in statement.targets:
                if isinstance(target, ast.Name) and target.id in ('VERSION', 'DATABASE', 'ROLE'):
                    values[target.id] = ast.literal_eval(statement.value)
    need(set(values) == {'VERSION', 'DATABASE', 'ROLE'} and values['VERSION'] == 'ksf.unified-db.v1'
         and all(isinstance(v, str) and v for v in values.values()), 'UNRECOGNIZED_RUNTIME_CONSTANTS')
    return {'database': values['DATABASE'], 'role': values['ROLE'],
            'moduleSha256': hashlib.sha256(raw).hexdigest()}


def identity(connection, connector, database=None):
    pg = None
    try:
        kwargs = dict(connection)
        # The service read observes its actual database without substituting the package name.
        if database is not None:
            kwargs['dbname'] = database
        kwargs.update(connect_timeout=10, application_name='ksf-unified-target-readonly-probe',
                      options='-c default_transaction_read_only=on')
        pg = connector(**kwargs)
        row = pg.execute("SELECT current_database(),current_user,session_user,"
                         "current_setting('transaction_read_only'),current_setting('server_version'),"
                         "(SELECT count(*) FROM pg_namespace WHERE nspname='ksf_voc_control' "
                         "OR left(nspname,5)='ksfv_')").fetchone()
        need(row[3] == 'on', 'READ_ONLY_TRANSACTION_REQUIRED')
        return {'ok': True, 'database': row[0], 'role': row[1], 'sessionRole': row[2],
                'transactionReadOnly': True, 'serverVersion': row[4], 'unifiedSchemaCount': row[5]}
    except Exception:
        # Database errors can contain DSNs or passwords; never return their original messages.
        return {'ok': False, 'errorCode': 'READ_ONLY_IDENTITY_QUERY_FAILED'}
    finally:
        if pg is not None:
            pg.rollback()
            pg.close()


def run(config, root, connector=None):
    value = private_configuration(config)
    expected = runtime_identity(root)
    if connector is None:
        import psycopg
        connector = psycopg.connect
    service = identity(value['connection'], connector)
    # This is the exact database override performed by the sealed migration runtime.
    runtime = identity(value['connection'], connector, expected['database'])
    result = {'schema': 'ksf.unified-target-readonly-probe.v1', 'ok': service['ok'] and runtime['ok'],
              'configuredDatabase': value['database'], 'packageExpected': expected,
              'serviceIdentity': service, 'runtimeConnectionIdentity': runtime,
              'configMatchesPackage': value['database'] == expected['database'],
              'readOnly': True, 'databaseWrites': 0, 'configurationModified': False,
              'migrationGuardModified': False, 'migrationExecuted': False}
    for label, actual in (('serviceMatchesPackage', service), ('runtimeIdentityMatchesPackage', runtime)):
        result[label] = actual['ok'] and (actual['database'], actual['role']) == (expected['database'], expected['role'])
    result['targetIdentityVerified'] = result['ok'] and result['configMatchesPackage'] and result['serviceMatchesPackage'] and result['runtimeIdentityMatchesPackage']
    return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--config', required=True, type=Path)
    parser.add_argument('--runtime-root', required=True, type=Path)
    args = parser.parse_args()
    try:
        result = run(args.config, args.runtime_root)
    except Exception:
        result = {'ok': False, 'errorCode': 'TARGET_PROBE_PRECONDITION_FAILED',
                  'readOnly': True, 'databaseWrites': 0, 'migrationExecuted': False}
    print(json.dumps(result, ensure_ascii=True, indent=2))
    return 0 if result.get('ok') else 1


if __name__ == '__main__':
    sys.exit(main())