187 lines
9.0 KiB
Python
187 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Build an independent evidence bundle. Never infer business PASS from tool exit codes."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import zipfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path, PureWindowsPath
|
|
|
|
VERDICTS = {'passed', 'failed', 'blocked', 'inconclusive'}
|
|
|
|
|
|
def digest(path: Path) -> str:
|
|
h = hashlib.sha256()
|
|
with path.open('rb') as stream:
|
|
for block in iter(lambda: stream.read(1024 * 1024), b''):
|
|
h.update(block)
|
|
return h.hexdigest()
|
|
|
|
|
|
def inside(root: Path, name: str) -> Path:
|
|
if not isinstance(name, str) or not name.strip():
|
|
raise ValueError('Evidence path must be a non-empty relative path')
|
|
if Path(name).is_absolute() or PureWindowsPath(name).is_absolute() or ':' in name:
|
|
raise ValueError('Evidence path must be relative to evidence_root')
|
|
candidate = (root / name).resolve()
|
|
if not candidate.is_relative_to(root) or '..' in Path(name.replace('\\', '/')).parts:
|
|
raise ValueError('Evidence path escapes evidence_root')
|
|
return candidate
|
|
|
|
|
|
def overall(assertions: list[dict]) -> str:
|
|
required = [a['verdict'] for a in assertions if a.get('required', True)]
|
|
if not required:
|
|
return 'inconclusive'
|
|
for status in ('failed', 'blocked', 'inconclusive'):
|
|
if status in required:
|
|
return status
|
|
return 'passed'
|
|
|
|
|
|
def cell(value) -> str:
|
|
return str(value).replace('|', '\\|').replace('\n', '<br>')
|
|
|
|
|
|
def bundle(manifest_path: Path, output: Path) -> dict:
|
|
manifest_path = manifest_path.resolve(strict=True)
|
|
output = output.resolve()
|
|
if output.exists():
|
|
raise FileExistsError('Output already exists; use a new run directory')
|
|
source = json.loads(manifest_path.read_text(encoding='utf-8-sig'))
|
|
if not isinstance(source, dict) or source.get('schema_version') != 1:
|
|
raise ValueError('schema_version must be 1')
|
|
run_id = source.get('run_id')
|
|
if not isinstance(run_id, str) or not re.fullmatch(r'[A-Za-z0-9_.-]{1,100}', run_id):
|
|
raise ValueError('run_id must be 1..100 letters, digits, _, -, or .')
|
|
root = (manifest_path.parent / source.get('evidence_root', '.')).resolve(strict=True)
|
|
assertions = source.get('assertions')
|
|
if not root.is_dir() or not isinstance(assertions, list):
|
|
raise ValueError('evidence_root must be a directory and assertions must be a list')
|
|
context = source.get('context', {})
|
|
if not isinstance(context, dict):
|
|
raise ValueError('context must be an object')
|
|
execution_mode = context.get('execution_mode', 'unspecified')
|
|
if execution_mode not in {'live', 'offline', 'simulation', 'unspecified'}:
|
|
raise ValueError('execution_mode must be live, offline, simulation or unspecified')
|
|
real_device = context.get('real_device_test_executed')
|
|
if real_device is not None and not isinstance(real_device, bool):
|
|
raise ValueError('real_device_test_executed must be Boolean')
|
|
if execution_mode in {'offline', 'simulation'} and real_device:
|
|
raise ValueError('Offline/simulation cannot claim real device execution')
|
|
ids = set()
|
|
checked = []
|
|
artifacts = {}
|
|
for original in assertions:
|
|
if not isinstance(original, dict):
|
|
raise ValueError('Each assertion must be an object')
|
|
a = dict(original)
|
|
identity = a.get('id')
|
|
if not isinstance(identity, str) or not identity or identity in ids:
|
|
raise ValueError('Assertion IDs must be non-empty and unique')
|
|
ids.add(identity)
|
|
if a.get('verdict') not in VERDICTS or not isinstance(a.get('required', True), bool):
|
|
raise ValueError('Use explicit verdict and Boolean required')
|
|
if 'expected' not in a or 'actual' not in a:
|
|
raise ValueError('Each assertion needs expected and actual observations')
|
|
refs = a.get('evidence', [])
|
|
if not isinstance(refs, list):
|
|
raise ValueError('evidence must be a list')
|
|
a['requested_evidence'] = refs
|
|
problems = []
|
|
normalized = []
|
|
for ref in refs:
|
|
ref = {'path': ref} if isinstance(ref, str) else ref
|
|
if not isinstance(ref, dict):
|
|
raise ValueError('Evidence references must be paths or objects')
|
|
file = inside(root, ref.get('path'))
|
|
relative = file.relative_to(root).as_posix()
|
|
if not file.is_file():
|
|
problems.append(f'Missing evidence: {relative}')
|
|
continue
|
|
sha = digest(file)
|
|
expected_hash = ref.get('sha256')
|
|
artifact = artifacts.setdefault(relative, {'path': 'evidence/' + relative,
|
|
'sha256': sha, 'size_bytes': file.stat().st_size, 'integrity': 'recorded'})
|
|
if expected_hash is not None and expected_hash != sha:
|
|
problems.append(f'Hash mismatch: {relative}')
|
|
artifact['integrity'] = 'mismatch'
|
|
elif expected_hash is not None and artifact['integrity'] != 'mismatch':
|
|
artifact['integrity'] = 'verified'
|
|
normalized.append('evidence/' + relative)
|
|
a['requested_verdict'] = a['verdict']
|
|
a['evidence'] = normalized
|
|
if a['verdict'] in {'passed', 'failed'} and (problems or not normalized):
|
|
a['verdict'] = 'inconclusive'
|
|
if not normalized:
|
|
problems.append('No valid evidence for the asserted observation')
|
|
a['evidence_issues'] = problems
|
|
checked.append(a)
|
|
result = {
|
|
'schema_version': 1, 'run_id': run_id,
|
|
'created_at': datetime.now(timezone.utc).isoformat(),
|
|
'report_type': 'independent_combined_evidence',
|
|
'verdict': overall(checked), 'assertions': checked,
|
|
'artifacts': list(artifacts.values()),
|
|
'context': context,
|
|
'notice': 'Evidence integrity is checked; the author remains responsible for semantic assertions. '
|
|
'Original artifacts may contain internal data. This is not an Artemis native verdict.',
|
|
}
|
|
# All input validation precedes output creation. A changed source aborts rather than publishing stale hashes.
|
|
output.mkdir(parents=True, exist_ok=False)
|
|
for relative, artifact in artifacts.items():
|
|
target = output / artifact['path']
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(root / relative, target)
|
|
if digest(target) != artifact['sha256']:
|
|
raise ValueError('Evidence changed while copying; output is incomplete')
|
|
(output / 'manifest.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
live_label = '是' if real_device is True else ('否' if real_device is False else '未标注')
|
|
lines = [f'# 联合测试记录：{run_id}', '', f"结论：**{result['verdict']}**", '',
|
|
f'运行性质：**{execution_mode}**；含真机操作：**{live_label}**。', '',
|
|
'此报告关联 PC 与手机证据，不改写 Artemis 原生判定。工具执行结束不等于测试通过。', '',
|
|
'| 断言 | 必需 | 预期 | 实际 | 判定 | 证据 |', '| --- | --- | --- | --- | --- | --- |']
|
|
for a in checked:
|
|
refs_text = ', '.join(f'[{Path(p).name}]({p.replace(" ", "%20")})' for p in a['evidence'])
|
|
lines.append('| ' + ' | '.join(cell(v) for v in (
|
|
a['id'], a.get('required', True), a['expected'], a['actual'], a['verdict'], refs_text)) + ' |')
|
|
for problem in a['evidence_issues']:
|
|
lines.append(f"\n证据问题（{cell(a['id'])}）：{cell(problem)}\n")
|
|
lines += ['', '机器可读的运行关联信息、文件哈希及原始请求判定见 manifest.json。',
|
|
'归档包含原始证据，按内部测试资料保管；本脚本不自动向外部发送或发布。', '']
|
|
(output / 'report.md').write_text('\n'.join(lines), encoding='utf-8')
|
|
with zipfile.ZipFile(output / 'bundle.zip', 'x', zipfile.ZIP_DEFLATED) as archive:
|
|
for file in sorted(output.rglob('*')):
|
|
if file.is_file() and file != output / 'bundle.zip':
|
|
archive.write(file, file.relative_to(output).as_posix())
|
|
return result
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
sub = parser.add_subparsers(dest='command', required=True)
|
|
b = sub.add_parser('bundle')
|
|
b.add_argument('--manifest', type=Path, required=True)
|
|
b.add_argument('--out', type=Path, required=True)
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
result = bundle(args.manifest, args.out)
|
|
print(json.dumps({'execution_status': 'completed', 'verdict': result['verdict'],
|
|
'output': str(args.out.resolve()), 'assertions': len(result['assertions'])},
|
|
ensure_ascii=False))
|
|
return 0
|
|
except (OSError, ValueError, TypeError) as exc:
|
|
print(json.dumps({'execution_status': 'error', 'message': str(exc)}, ensure_ascii=False))
|
|
return 2
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if hasattr(sys.stdout, 'reconfigure'):
|
|
sys.stdout.reconfigure(encoding='utf-8')
|
|
raise SystemExit(main())
|