"""生图接口指纹与性能测试（OpenAI Images API 兼容接口） 对每张出图检查： - HTTP 状态、首包耗时（≈生成耗时）、总耗时（含下行传输）、响应大小、响应头中的请求 ID - usage（output/input tokens） - 像素尺寸是否与请求一致 - PNG 块结构、IDAT 分块情况（单块 / 32768 字节分块） - 是否带 C2PA（caBX 块）、softwareAgent、claim generator、签名方字符串 - C2PA 数据哈希与文件字节是否一致（不一致说明签名后图片被改过） 参考指纹： OpenAI API 原签 softwareAgent = API/gpt-image，IHDR/caBX/IDAT/IEND，单块 IDAT ChatGPT 网页版 softwareAgent = ChatGPT/gpt-image，IDAT 按 32768 字节分块 Azure OpenAI softwareAgent = Azure OpenAI ImageGen，签发方 Microsoft 无 C2PA IHDR/IDAT/IEND，来源无法验证 只用 Python 3 标准库，不需要安装依赖。 用法： # 先看会发哪些请求（不花钱） python image_fingerprint_test.py --base https://api.example.com --model gpt-image-2 --dry-run # 正式测试，Key 从环境变量读取 set IMAGE_API_KEY=sk-... (Windows) export IMAGE_API_KEY=sk-... (Linux/macOS) python image_fingerprint_test.py --base https://api.example.com --model gpt-image-2 \\ --edit-image src.png --out run1 # gpt-image-2.5 系列可加更多质量档 python image_fingerprint_test.py --base ... --model gpt-image-2.5-flare --qualities low,medium,high,xhigh,max,auto # 只分析本地已有的图片 python image_fingerprint_test.py --inspect a.png b.png 注意：每个用例都会真实出图计费，4K high 单张较贵。先用 --dry-run 确认用例数。 """ import argparse, base64, hashlib, json, os, re, struct, sys, time, urllib.error, urllib.request, uuid from concurrent.futures import ThreadPoolExecutor from pathlib import Path PROMPTS = { 'circuit': 'Macro photo of a green printed circuit board, dense tiny SMD components, fine copper traces and small white silkscreen labels, sharp focus.', 'poster': 'A minimalist concert poster with large bold title text "NIGHT SIGNAL" and small Chinese subtitle "夜间信号", dark blue background.', 'edit': 'Change the background to deep red. Keep all text, layout and typography unchanged.', } def build_cases(qualities, edit_image, repeat): cases = [] # (name, kind, size, quality, response_format, prompt) for r in range(1, repeat + 1): sfx = f'_{r}' if repeat > 1 else '' cases += [(f'size_circuit_720_high{sfx}', 'gen', '1280x720', 'high', 'b64_json', 'circuit'), (f'size_circuit_2k_high{sfx}', 'gen', '2048x1152', 'high', 'b64_json', 'circuit'), (f'size_poster_4k_high{sfx}', 'gen', '3840x2160', 'high', 'b64_json', 'poster')] cases += [(f'q_{q}_2k{sfx}', 'gen', '2048x1152', q, 'b64_json', 'circuit') for q in qualities if q != 'high'] if edit_image: cases.append((f'edit_2k_high{sfx}', 'edit', '2048x1152', 'high', 'b64_json', 'edit')) cases.append((f'url_low_720{sfx}', 'gen', '1280x720', 'low', 'url', 'poster')) return cases def post(base, key, kind, body, edit_bytes, timeout): """Returns status, body, headers, seconds to response headers, seconds total.""" h = {'Authorization': 'Bearer ' + key, 'User-Agent': 'image-fingerprint-test/1.0'} if kind == 'gen': data = json.dumps(body, ensure_ascii=False).encode('utf-8') h['Content-Type'] = 'application/json' url = base + '/v1/images/generations' else: bd = uuid.uuid4().hex parts = [f'--{bd}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode('utf-8') for k, v in body.items()] parts.append(f'--{bd}\r\nContent-Disposition: form-data; name="image"; filename="src.png"\r\nContent-Type: image/png\r\n\r\n'.encode('utf-8') + edit_bytes + b'\r\n') parts.append(f'--{bd}--\r\n'.encode('utf-8')) data = b''.join(parts) h['Content-Type'] = 'multipart/form-data; boundary=' + bd url = base + '/v1/images/edits' req = urllib.request.Request(url, data=data, headers=h, method='POST') t = time.time() try: r = urllib.request.urlopen(req, timeout=timeout) except urllib.error.HTTPError as e: th = time.time() - t return e.code, e.read(), dict(e.headers), th, time.time() - t with r: th = time.time() - t raw = r.read() return r.status, raw, dict(r.headers), th, time.time() - t # ---- C2PA / PNG analysis ---- def cbor_text(b, i): """Reads a CBOR text string at b[i]; returns (text, next index) or (None, i).""" if i >= len(b): return None, i t = b[i] if 0x60 <= t <= 0x77: n, j = t - 0x60, i + 1 elif t == 0x78 and i + 1 < len(b): n, j = b[i + 1], i + 2 elif t == 0x79 and i + 2 < len(b): n, j = struct.unpack('>H', b[i + 1:i + 3])[0], i + 3 else: return None, i return b[j:j + n].decode('utf-8', 'replace'), j + n def cbor_label(b, key): """Finds a CBOR text key and returns its value: a text, or name/version joined from a small map.""" k = key.encode('utf-8') pos = b.find(bytes([0x60 + len(k)]) + k) if len(k) <= 23 else b.find(bytes([0x78, len(k)]) + k) if pos == -1: return None i = pos + 1 + len(k) + (0 if len(k) <= 23 else 1) s, _ = cbor_text(b, i) if s is not None: return s if i < len(b) and (0xa1 <= b[i] <= 0xb7 or b[i] == 0xbf): # map such as {name: "API", version: "gpt-image"}; 0xbf = indefinite length (Azure) n, j, vals = (b[i] - 0xa0 if b[i] != 0xbf else 8), i + 1, {} for _ in range(n): mk, j = cbor_text(b, j) mv, j2 = cbor_text(b, j) if mk is None or mv is None: break vals[mk], j = mv, j2 if 'name' in vals: return vals['name'] + ('/' + vals['version'] if 'version' in vals else '') return None def c2pa_hash_ok(b, start, length): box = b[start:start + length] i = box.find(b'c2pa.hash.data') if i == -1: return None cands, j = [], box.find(b'dhashX ', i) while j != -1 and j < i + 4000: cands.append(box[j + 7:j + 39]) j = box.find(b'dhashX ', j + 1) for s0, l0 in ((start, length), (start + 8, length - 12), (start + 8, length - 8)): if hashlib.sha256(b[:s0] + b[s0 + l0:]).digest() in cands: return True return False def classify(agent, has_c2pa): if not has_c2pa: return '无C2PA' a = (agent or '').lower() if a.startswith('api/'): return 'OpenAI API' if a.startswith('chatgpt'): return 'ChatGPT网页版' if 'azure' in a: return 'Azure OpenAI' return '其他:' + (agent or '?') def inspect_image(b): if not b.startswith(b'\x89PNG\r\n\x1a\n'): fmt = 'jpeg' if b[:3] == b'\xff\xd8\xff' else 'webp' if b[8:12] == b'WEBP' else b[:8].hex() c2pa = b'c2pa' in b return {'format': fmt, 'c2pa': c2pa, 'agent': cbor_label(b, 'softwareAgent') if c2pa else None, 'verdict': classify(cbor_label(b, 'softwareAgent'), c2pa), 'bytes': len(b)} pos, chunks, idat, cab, w = 8, [], [], None, None while pos + 8 <= len(b): n, typ = struct.unpack('>I4s', b[pos:pos + 8]) typ = typ.decode('latin1') if typ == 'IDAT': idat.append(n) if typ == 'caBX' and cab is None: cab = (pos, 12 + n) if typ == 'IHDR': w, h = struct.unpack('>II', b[pos + 8:pos + 16]) if not chunks or chunks[-1] != typ: chunks.append(typ) pos += 12 + n if typ == 'IEND': break region = b[cab[0]:cab[0] + cab[1]] if cab else b'' agent = cbor_label(region, 'softwareAgent') if cab else None gen = cbor_label(region, 'claim_generator') if cab else None if cab and not gen: k = region.find(b'claim_generator_info') gen = cbor_label(region[k:], 'name') if k != -1 else None signers = sorted({m.decode('latin1') for m in re.findall(rb'(?:OpenAI|Microsoft|Azure|Google|Adobe)(?: (?:[A-Z][a-z]+|[A-Z]{2,}))*', region)})[:6] return {'format': 'png', 'w': w, 'h': h if w else None, 'chunks': '/'.join(chunks), 'idat_n': len(idat), 'idat_max': max(idat) if idat else 0, 'c2pa': cab is not None, 'agent': agent, 'claim_generator': gen, 'signer_strings': signers, 'hash_ok': c2pa_hash_ok(b, *cab) if cab else None, 'verdict': classify(agent, cab is not None), 'bytes': len(b), 'sha256_12': hashlib.sha256(b).hexdigest()[:12]} # ---- run ---- def run_case(case, a, key, edit_bytes, outdir): name, kind, size, q, fmt, pk = case body = {'model': a.model, 'prompt': PROMPTS[pk], 'size': size, 'quality': q, 'n': 1, 'response_format': fmt} code, raw, hdr, t_head, t_total = post(a.base, key, kind, body, edit_bytes, a.timeout) out = {'case': name, 'size_req': size, 'quality': q, 'format_req': fmt, 'http': code, 'sec_headers': round(t_head, 1), 'sec_total': round(t_total, 1), 'resp_mb': round(len(raw) / 1e6, 2), 'request_ids': {k: v for k, v in hdr.items() if 'request' in k.lower() or 'trace' in k.lower()}} try: d = json.loads(raw) except ValueError: out['error'] = raw[:300].decode('utf-8', 'replace') return out if code != 200: out['error'] = json.dumps(d, ensure_ascii=False)[:300] return out item = (d.get('data') or [{}])[0] u = d.get('usage') or {} out['usage'] = {'output_tokens': u.get('output_tokens'), 'input_tokens': u.get('input_tokens'), 'output_details': u.get('output_tokens_details')} out['created'] = d.get('created') if item.get('b64_json'): img = base64.b64decode(item['b64_json']) elif item.get('url'): out['url_host'] = item['url'].split('/')[2] t = time.time() with urllib.request.urlopen(urllib.request.Request(item['url'], headers={'User-Agent': 'image-fingerprint-test/1.0'}), timeout=a.timeout) as r: img = r.read() out['url_fetch_sec'] = round(time.time() - t, 1) else: out['error'] = 'no image in response' return out (outdir / (name + '.png')).write_bytes(img) out.update(inspect_image(img)) out['size_hit'] = out.get('w') is not None and f"{out['w']}x{out['h']}" == size return out def print_table(rows): cols = ['case', 'http', 'sec_headers', 'sec_total', 'resp_mb', 'verdict', 'agent', 'chunks', 'idat_n', 'hash_ok', 'size_hit'] print('\n' + ' | '.join(cols)) for r in rows: tok = (r.get('usage') or {}).get('output_tokens') print(' | '.join(str(r.get(c, '')) for c in cols) + f' | tokens={tok}' + (f" | {r['error'][:80]}" if r.get('error') else '')) ok = [r for r in rows if r.get('http') == 200 and 'verdict' in r] if ok: dist = {} for r in ok: dist[r['verdict']] = dist.get(r['verdict'], 0) + 1 print('\n指纹分布:', ', '.join(f'{k} {v}/{len(ok)}' for k, v in dist.items())) bad = [r['case'] for r in ok if r.get('hash_ok') is False] if bad: print('C2PA 哈希不一致（签名后被改动）:', ', '.join(bad)) def main(): p = argparse.ArgumentParser(description='生图接口指纹与性能测试') p.add_argument('--base', help='接口根地址，如 https://api.example.com（不含 /v1）') p.add_argument('--key', help='API Key；不填则读环境变量 IMAGE_API_KEY') p.add_argument('--model', default='gpt-image-2') p.add_argument('--qualities', default='low,medium,high,auto', help='2K 质量档，逗号分隔') p.add_argument('--edit-image', help='编辑用例的输入 PNG；不填则跳过编辑用例') p.add_argument('--repeat', type=int, default=1, help='整组用例重复次数') p.add_argument('--concurrency', type=int, default=4) p.add_argument('--timeout', type=int, default=900) p.add_argument('--out', default='fingerprint_run', help='输出目录（图片 + result.json）') p.add_argument('--dry-run', action='store_true', help='只列出用例，不发请求') p.add_argument('--inspect', nargs='+', metavar='FILE', help='只分析本地图片') a = p.parse_args() if a.inspect: for f in a.inspect: print(f, json.dumps(inspect_image(Path(f).read_bytes()), ensure_ascii=False)) return if not a.base: p.error('需要 --base') a.base = a.base.rstrip('/') if a.base.endswith('/v1'): a.base = a.base[:-3] edit_bytes = Path(a.edit_image).read_bytes() if a.edit_image else None cases = build_cases([q.strip() for q in a.qualities.split(',') if q.strip()], edit_bytes is not None, a.repeat) print(f'{len(cases)} 个用例，模型 {a.model}，地址 {a.base}') for c in cases: print(' ', c[0], c[1], c[2], c[3], c[4]) if a.dry_run: return key = a.key or os.environ.get('IMAGE_API_KEY') if not key: p.error('需要 --key 或环境变量 IMAGE_API_KEY') outdir = Path(a.out) outdir.mkdir(parents=True, exist_ok=True) rows = [] with ThreadPoolExecutor(a.concurrency) as ex: for r in ex.map(lambda c: run_case(c, a, key, edit_bytes, outdir), cases): rows.append(r) print(json.dumps(r, ensure_ascii=False), flush=True) meta = {'base': a.base, 'model': a.model, 'started_note': time.strftime('%Y-%m-%d %H:%M:%S %z')} (outdir / 'result.json').write_text(json.dumps({'meta': meta, 'cases': rows}, ensure_ascii=False, indent=1), encoding='utf-8') print_table(rows) print(f'\n图片和 result.json 在 {outdir.resolve()}') if __name__ == '__main__': sys.exit(main())