企业多款 App 接入加固：先建立能识别候选、签名和渠道的资产台账

import json
import re
import sys
from collections import defaultdict
from pathlib import Path

REQUIRED = {
    'assetId',
    'platform',
    'applicationId',
    'variant',
    'signingFingerprint',
    'artifactSha256',
    'channel',
    'owner',
    'status',
}
FORBIDDEN = {'privateKey', 'password', 'token', 'customerData'}
PLATFORMS = {'android', 'ios'}
STATUSES = {'current', 'historical', 'blocked', 'retired'}
SHA256 = re.compile(r'^[a-f0-9]{64}$')
AMBIGUOUS = re.compile(r'^(both|joint|tbd|双方|共同负责|待定)$', re.I)

def stop(message):
    raise SystemExit(message)

def required_text(record, field, index):
    value = str(record[field]).strip()
    if not value:
        stop(f'record {index} has empty {field}')
    return value

def validate(record, index):
    if not isinstance(record, dict):
        stop(f'record {index} is not an object')
    if FORBIDDEN.intersection(record):
        stop(f'record {index} contains forbidden fields')
    missing = sorted(REQUIRED - record.keys())
    if missing:
        stop(f'record {index} lacks fields: {missing}')
    if record['platform'] not in PLATFORMS:
        stop(f'record {index} has invalid platform')
    if record['status'] not in STATUSES:
        stop(f'record {index} has invalid status')
    digest = required_text(record, 'artifactSha256', index)
    if not SHA256.fullmatch(digest):
        stop(f'record {index} has invalid artifact digest')
    owner = required_text(record, 'owner', index)
    if AMBIGUOUS.fullmatch(owner):
        stop(f'record {index} has ambiguous owner')
    identity = (
        record['platform'],
        required_text(record, 'applicationId', index),
        required_text(record, 'variant', index),
        required_text(record, 'signingFingerprint', index),
        required_text(record, 'channel', index),
    )
    return required_text(record, 'assetId', index), digest, identity, record['status']

if len(sys.argv) != 2:
    stop('usage: validate_asset_register.py register.json')
source = Path(sys.argv[1])
if not source.is_file():
    raise SystemExit('asset register file is missing')
data = json.loads(source.read_text(encoding='utf-8'))
records = data.get('assets') if isinstance(data, dict) else None
if not isinstance(records, list) or not records:
    raise SystemExit('assets array is missing')
validated = [validate(record, index) for index, record in enumerate(records)]
asset_ids = [item[0] for item in validated]
if len(asset_ids) != len(set(asset_ids)):
    raise SystemExit('assetId is not unique')
digest_identities = defaultdict(set)
current_by_identity = defaultdict(set)
for asset_id, digest, identity, status in validated:
    digest_identities[digest].add(identity)
    if status == 'current':
        current_by_identity[identity].add(digest)
conflicting_digests = [digest for digest, identities in digest_identities.items() if len(identities) > 1]
if conflicting_digests:
    raise SystemExit('one artifact digest maps to multiple identities')
conflicting_current = [identity for identity, digests in current_by_identity.items() if len(digests) > 1]
if conflicting_current:
    raise SystemExit('one identity has multiple current candidates')
print(json.dumps({'assets': len(records), 'status': 'pass'}))