ute2e/tools/verify-scene-artifacts.py

"""Scan this scene report only. Never print matched credentials or request data."""
from pathlib import Path
from collections import Counter
from html.parser import HTMLParser
from urllib.parse import unquote, urlsplit
import argparse
import base64
import datetime
import hashlib
import html
import io
import json
import re
import subprocess
import zipfile

from PIL import Image

WORKSPACE = next(parent for parent in Path(__file__).resolve().parents if (parent / 'unreal_tran/ute2e').is_dir())
ROOT = WORKSPACE / 'unreal_tran'
REPORT = ROOT / 'ute2e/reports/scene-editor-closure-20260906'
OUTPUT = REPORT / 'artifact-checks.json'
FINAL_FILES = ('results.json', 'validation.json', 'cleanup.json', 'playwright/index.html')
TEXT_SUFFIXES = {'.ts', '.js', '.mjs', '.cjs', '.py', '.md', '.json', '.html', '.css', '.txt', '.log', '.trace', '.network', '.svg', '.xml', '.yml', '.yaml', '.csv', '.vue', '.sql'}
IMAGES = {'.png', '.jpg', '.jpeg', '.webp'}
JWT = re.compile(r'(?<![A-Za-z0-9_-])(eyJ[A-Za-z0-9_-]{5,})\.([A-Za-z0-9_-]{8,})\.([A-Za-z0-9_-]{8,})(?![A-Za-z0-9_-])')
ZIP_DATA = re.compile(r'data:application/zip;base64,([A-Za-z0-9+/=\r\n]+)')
PRIVATE_PATH = re.compile(r'(?:^|/)(?:账号\.md|credentials?(?:\.[^/]+)?|storage[-_]?state\.json|fixtures\.json|\.env(?:\.[^/]+)?|[^/]*\.sql\.gz)$', re.I)
SECRET_KEY = re.compile(r'^(?:password|passwd|pwd|accessToken|refreshToken|apiKey|secret)$', re.I)
ASSIGNMENT = re.compile(r'(?:password|passwd|pwd|secret|apiKey|accessToken|refreshToken|authorization|cookie|密码)[\s"\x27]*[:=][\s"\x27]*[^,;\r\n]{0,40}$', re.I)


def decode(data):
    return data.decode('utf-16' if data[:2] in (b'\xff\xfe', b'\xfe\xff') else 'utf-8-sig', errors='replace').replace('\r\n', '\n')


def credentials():
    lines = [line.strip() for line in (ROOT / '账号.MD').read_text(encoding='utf-8-sig').splitlines() if line.strip()]
    values = {re.split(r'[:：]', line, maxsplit=1)[-1].strip() for line in lines[1::2]}
    fixture = WORKSPACE / '.codex-tmp/scene-closure-20260906/fixtures.json'
    def collect(value):
        if isinstance(value, dict):
            for key, child in value.items():
                if SECRET_KEY.fullmatch(key) and isinstance(child, str) and child:
                    values.add(child)
                elif isinstance(child, (dict, list)):
                    collect(child)
        elif isinstance(value, list):
            for child in value:
                collect(child)
    if not fixture.is_file():
        raise ValueError('Missing private reference')
    collect(json.loads(fixture.read_text(encoding='utf-8-sig')))
    if not values or '' in values:
        raise ValueError('Missing credential values')
    return sorted(values, key=len, reverse=True)


def decoded_jwts(text):
    for match in JWT.finditer(text):
        try:
            head = json.loads(base64.urlsafe_b64decode(match[1] + '=' * (-len(match[1]) % 4)))
            body = json.loads(base64.urlsafe_b64decode(match[2] + '=' * (-len(match[2]) % 4)))
            if isinstance(head, dict) and isinstance(body, dict) and ('alg' in head or str(head.get('typ', '')).upper() == 'JWT'):
                yield match
        except (ValueError, UnicodeError):
            continue


def text_matches(text, values, baselines=()):
    findings, exemptions = [], []
    for value in values:
        for needle in {value, json.dumps(value, ensure_ascii=True)[1:-1], html.escape(value, quote=True)}:
            used = Counter()
            for match in re.finditer(re.escape(needle), text):
                offset = match.start()
                prefix = text[max(text.rfind('\n', 0, offset) + 1, offset - 100):offset]
                context = text[max(0, offset - 80):min(len(text), match.end() + 80)]
                accepted = None
                # Only the caller-supplied, committed Playwright static shell is
                # eligible. An assignment or decoded ZIP can never use this waiver.
                if not ASSIGNMENT.search(prefix) and len(context) >= 64:
                    for name, previous in baselines:
                        key = (name, context)
                        if context in previous and used[key] < previous.count(context) * context.count(needle):
                            accepted = name
                            used[key] += 1
                            break
                if accepted:
                    exemptions.append({'rule': 'unchanged-playwright-vendor-context', 'baseline': accepted})
                else:
                    findings.append({'rule': 'known-credential', 'line': text.count('\n', 0, offset) + 1})
    findings.extend({'rule': 'decodable-jwt', 'line': text.count('\n', 0, match.start()) + 1} for match in decoded_jwts(text))
    return findings, exemptions


def status_pass(value):
    return str(value or '').upper() in ('PASS', 'PASSED', 'SUCCESS')


def final_input_state(documents, requested):
    reasons = []
    if not requested:
        reasons.append('final-scan-not-requested')
    for name in FINAL_FILES:
        if name not in documents:
            reasons.append('missing-final-input:' + name)
    validation = documents.get('validation.json', {})
    if not (validation.get('finalRun') is True or validation.get('e2eComplete') is True or validation.get('e2e', {}).get('complete') is True):
        reasons.append('final-run-not-confirmed')
    if not status_pass(validation.get('overallStatus', validation.get('status'))):
        reasons.append('validation-not-pass')
    if not status_pass(documents.get('cleanup.json', {}).get('status')):
        reasons.append('cleanup-not-pass')
    tests = []
    def visit(suites):
        for suite in suites:
            for spec in suite.get('specs', []):
                tests.extend(spec.get('tests', []))
            visit(suite.get('suites', []))
    results = documents.get('results.json', {})
    visit(results.get('suites', []))
    if not tests or any(test.get('status') in ('unexpected', 'flaky', 'skipped') or not test.get('results') or test['results'][-1].get('status') != 'passed' for test in tests):
        reasons.append('current-e2e-not-all-pass')
    if results.get('errors') or any(results.get('stats', {}).get(key, 0) for key in ('unexpected', 'flaky', 'skipped')):
        reasons.append('current-e2e-has-errors')
    return reasons


class LocalLinks(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.links = []
    def handle_starttag(self, tag, attrs):
        self.links.extend(value for key, value in attrs if key in ('href', 'src', 'poster') and value)


def resolve_local_link(source, value, root):
    parsed = urlsplit(value)
    if parsed.scheme in ('http', 'https', 'data', 'mailto', 'tel', 'javascript', 'blob') or parsed.netloc or not parsed.path:
        return None
    if parsed.scheme:
        raise ValueError('Unsupported local URL scheme')
    target = (source.parent / unquote(parsed.path)).resolve()
    if not target.is_relative_to(root):
        raise ValueError('Local URL outside workspace')
    return target


def image_metadata(payload):
    with Image.open(io.BytesIO(payload)) as img:
        img.verify()
    with Image.open(io.BytesIO(payload)) as img:
        img.load()
        if min(img.size) <= 0:
            raise ValueError('Empty image')
        values = list(img.info.values()) + list(img.getexif().values())
        metadata = '\n'.join(decode(value) if isinstance(value, bytes) else str(value) for value in values if isinstance(value, (str, bytes)))
        return img.format, img.size, metadata


def self_checks():
    value = 'test-only-scene-scanner-value'
    vendor = 'const alphabet="ABCDEFGHIJKLMNOPQRSTUVWXYZ-prefix-' + value + '-suffix-abcdefghijklmnopqrstuvwxyz-0123456789";'
    assert not text_matches(vendor, [value], [('HEAD:vendor', vendor)])[0]
    assert text_matches('password="' + value + '"', [value], [('HEAD:vendor', 'password="' + value + '"')])[0]
    assert text_matches(vendor, [value])[0]
    enc = lambda obj: base64.urlsafe_b64encode(json.dumps(obj).encode()).decode().rstrip('=')
    token = enc({'alg': 'HS256'}) + '.' + enc({'sub': 'scanner-test'}) + '.testsigsample'
    assert text_matches(token, [])[0]
    assert not text_matches('Bearer ${token}; eyJnot-a-token', [])[0]
    archive = io.BytesIO()
    with zipfile.ZipFile(archive, 'w') as item:
        item.writestr('report.json', json.dumps({'accessToken': token}))
    embedded = 'data:application/zip;base64,' + base64.b64encode(archive.getvalue()).decode()
    with zipfile.ZipFile(io.BytesIO(base64.b64decode(ZIP_DATA.search(embedded)[1]))) as item:
        assert text_matches(decode(item.read('report.json')), [])[0]
    png = io.BytesIO()
    Image.new('RGB', (2, 3)).save(png, format='PNG')
    assert image_metadata(png.getvalue())[:2] == ('PNG', (2, 3))
    try:
        image_metadata(png.getvalue()[:20])
    except Exception:
        pass
    else:
        raise AssertionError('Truncated image was accepted')
    docs = {'results.json': {'suites': [{'specs': [{'tests': [{'status': 'expected', 'results': [{'status': 'passed'}]}]}]}]}, 'validation.json': {'status': 'PASS', 'e2eComplete': True}, 'cleanup.json': {'status': 'PASS'}, 'playwright/index.html': True}
    assert not final_input_state(docs, True)
    assert final_input_state(docs, False)
    assert final_input_state({}, True)
    docs['results.json']['suites'][0]['specs'][0]['tests'][0]['status'] = 'flaky'
    assert final_input_state(docs, True)
    source = WORKSPACE / 'example/index.html'
    assert resolve_local_link(source, '../%E6%96%87%E6%A1%A3/sample.pdf#x', WORKSPACE) == WORKSPACE / '文档/sample.pdf'
    assert resolve_local_link(source, 'https://example.test/image.png', WORKSPACE) is None
    try:
        resolve_local_link(source, '../../../outside.txt', WORKSPACE)
    except ValueError:
        pass
    else:
        raise AssertionError('Escaping local link was accepted')
    parser = LocalLinks()
    parser.feed('<img src="screenshots/test.png"><a href="data/evidence.json">x</a>')
    assert len(parser.links) == 2
    return 16


def scan(final=False):
    checks = self_checks()
    values = credentials()
    findings, exemptions, pending = [], [], []
    counts = Counter()
    documents, hashes = {}, {}
    baselines = []
    for source in ('reports/role-permissions-20260906/playwright/index.html', 'reports/model-editor-closure-20260906/playwright/index.html'):
        result = subprocess.run(['git', 'show', 'HEAD:' + source], cwd=ROOT / 'ute2e', stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False)
        if result.returncode == 0:
            baselines.append(('ute2e/HEAD:' + source, ZIP_DATA.sub('data:application/zip;base64,[decoded separately]', decode(result.stdout))))

    def safe(label):
        for value in values:
            label = label.replace(value, '[redacted]')
        return JWT.sub('[redacted-jwt]', label)

    def record(label, rule, **extra):
        findings.append({'file': safe(label), 'rule': rule, **extra})

    def scan_text(label, content, vendor=False, depth=0):
        counts['textArtifactsScanned'] += 1
        encoded = list(ZIP_DATA.finditer(content))
        plain = ZIP_DATA.sub('data:application/zip;base64,[decoded separately]', content)
        hits, waived = text_matches(plain, values, baselines if vendor else ())
        findings.extend({'file': safe(label), **hit} for hit in hits)
        exemptions.extend({'file': safe(label), **item} for item in waived)
        if 'playwrightReportBase64' in content and not encoded:
            record(label, 'playwright-embedded-metadata-missing')
        for index, match in enumerate(encoded):
            try:
                scan_zip(label + '!embedded-' + str(index + 1) + '.zip', base64.b64decode(match[1]), depth + 1)
            except Exception:
                record(label, 'embedded-archive-invalid')

    def scan_image(label, payload):
        try:
            _, _, metadata = image_metadata(payload)
            counts['imagesValidated'] += 1
            if metadata:
                scan_text(label + '!image-metadata', metadata)
        except Exception:
            record(label, 'image-validation-failed')

    def scan_zip(label, payload, depth=0):
        if depth > 3:
            record(label, 'archive-depth-exceeded')
            return
        try:
            with zipfile.ZipFile(io.BytesIO(payload)) as archive:
                entries = [entry for entry in archive.infolist() if not entry.is_dir()]
                if sum(entry.file_size for entry in entries) > 256 * 1024 * 1024:
                    record(label, 'archive-size-limit-exceeded')
                    return
                counts['archivesDecoded'] += 1
                for entry in entries:
                    child = label + '!' + entry.filename
                    if PRIVATE_PATH.search(entry.filename.replace('\\', '/')):
                        record(child, 'private-artifact-in-archive')
                    data = archive.read(entry)
                    counts['archiveEntriesScanned'] += 1
                    suffix = Path(entry.filename).suffix.lower()
                    if suffix == '.zip' or data[:4] == b'PK\x03\x04':
                        scan_zip(child, data, depth + 1)
                    elif suffix in IMAGES:
                        scan_image(child, data)
                    elif suffix in TEXT_SUFFIXES or b'\x00' not in data[:4096]:
                        scan_text(child, decode(data), depth=depth)
        except Exception:
            record(label, 'archive-read-failed')

    for file in sorted(REPORT.rglob('*')):
        if not file.is_file():
            continue
        label = file.relative_to(REPORT).as_posix()
        if file.is_symlink() or not file.resolve().is_relative_to(REPORT.resolve()):
            record(label, 'report-link-outside-scan-root')
            continue
        if PRIVATE_PATH.search(label):
            record(label, 'private-artifact-in-report')
        data = file.read_bytes()
        counts['filesScanned'] += 1
        if label in FINAL_FILES:
            hashes[label] = hashlib.sha256(data).hexdigest()
            try:
                documents[label] = json.loads(decode(data)) if label.endswith('.json') else True
            except Exception:
                record(label, 'final-input-invalid-json')
        suffix = file.suffix.lower()
        if suffix in IMAGES:
            scan_image(label, data)
        elif suffix == '.zip' or data[:4] == b'PK\x03\x04':
            scan_zip(label, data)
        elif suffix in TEXT_SUFFIXES or b'\x00' not in data[:4096]:
            content = decode(data)
            scan_text(label, content, vendor=label == 'playwright/index.html')
            if suffix == '.html':
                parser = LocalLinks()
                try:
                    parser.feed(content)
                    for value in parser.links:
                        target = resolve_local_link(file, value, WORKSPACE)
                        if target is not None:
                            counts['localLinksChecked'] += 1
                            if not target.exists():
                                record(label, 'missing-local-link', target=safe(str(target.relative_to(WORKSPACE)).replace('\\', '/')))
                except Exception:
                    record(label, 'invalid-html-local-link')
        else:
            counts['nontextBinaryFiles'] += 1
    pending.extend(final_input_state(documents, final))
    if not counts['filesScanned'] or not counts['imagesValidated']:
        pending.append('report-or-images-missing')
    status = 'FAIL' if findings else 'PENDING' if pending else 'PASS'
    output = {
        'time': datetime.datetime.now(datetime.timezone.utc).isoformat(),
        'status': status,
        'phase': 'final' if final else 'preliminary',
        'finalArtifactScan': final and not pending,
        'scope': '仅 scene-editor-closure-20260906 报告文件；不改应用，不调用 API，不清理数据。',
        'finalInputs': {name: name in documents for name in FINAL_FILES},
        'finalInputSha256': hashes,
        'pendingReasons': pending,
        'selfChecks': {'status': 'PASS', 'tests': checks},
        'counts': dict(counts),
        'findings': findings,
        'exemptions': {'count': len(exemptions), 'items': exemptions},
        'limitations': ['图片解码和文本元数据检查不代替人工视觉核验；未对图片像素进行 OCR。', '只校验 HTML 本地链接文件存在，不访问外部链接；Playwright 路由片段不作文件路径。', '只有当前最终输入完整且全部通过才可 PASS；已解码 ZIP 不适用静态第三方库基线豁免。']
    }
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(json.dumps(output, ensure_ascii=False, indent=2), encoding='utf-8')
    print(json.dumps({'status': status, 'finalArtifactScan': output['finalArtifactScan'], 'findings': len(findings), 'pending': pending, 'counts': dict(counts)}, ensure_ascii=False))
    return 0 if status == 'PASS' else 1 if status == 'FAIL' else 2


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--final', action='store_true')
    parser.add_argument('--self-test', action='store_true')
    args = parser.parse_args()
    if args.self_test:
        print(json.dumps({'status': 'PASS', 'tests': self_checks()}))
    else:
        try:
            raise SystemExit(scan(args.final))
        except Exception as error:
            # Do not include exception messages: they may contain source values.
            failure = {'time': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'status': 'FAIL', 'finalArtifactScan': False, 'finalInputSha256': {}, 'findings': [{'file': 'scanner', 'rule': 'scan-incomplete', 'errorType': type(error).__name__}]}
            OUTPUT.parent.mkdir(parents=True, exist_ok=True)
            OUTPUT.write_text(json.dumps(failure, indent=2), encoding='utf-8')
            print(json.dumps({'status': 'FAIL', 'errorType': type(error).__name__}))
            raise SystemExit(1)