415 lines
16 KiB
Python
415 lines
16 KiB
Python
"""First-keystroke forensic on real resume: Professional Experience vs +x.
|
|
|
|
Mirrors ParagraphEditor after editedRef=true:
|
|
- no origLines
|
|
- advances dropped when length != text (typing one char)
|
|
|
|
Compares per-glyph: char, gid, advance, origin, font resource, outline hash.
|
|
Does not modify engine logic beyond using the rebuilt binary.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import struct
|
|
import sys
|
|
import zlib
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "gateway"))
|
|
sys.path.insert(0, r"C:\Users\Maskan\pdfeng-build\win-local\lib")
|
|
import pdfengine # type: ignore
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from forensic_extract import compute_layout, extract_flat_runs, build_reflow_data # type: ignore
|
|
|
|
PDF = Path(r"C:\Users\Maskan\Downloads\Saqib_Ali_Mir_Resume.pdf")
|
|
OUT = Path(__file__).resolve().parent / "forensic_real"
|
|
FID = "Arial-BoldMT_TrueType_32"
|
|
TARGET = "Professional Experience"
|
|
TYPED = TARGET + "x"
|
|
TOL = 0.05
|
|
DPI = 288
|
|
|
|
|
|
def sha(b: bytes) -> str:
|
|
return hashlib.sha256(b).hexdigest()[:16]
|
|
|
|
|
|
def find_para(doc):
|
|
page = doc.get_page(0)
|
|
model = page.extract_document_model()
|
|
for idx, para in enumerate(model.paragraphs):
|
|
text = "".join(r.text or "" for ln in para.lines for r in ln.runs)
|
|
if TARGET in text or text.strip().startswith("Professional"):
|
|
return idx, para, text
|
|
raise RuntimeError("paragraph not found")
|
|
|
|
|
|
def collect_glyphs(para, want_prefix: str) -> list[dict]:
|
|
rows = []
|
|
for ln in para.lines:
|
|
for r in ln.runs:
|
|
glyphs = list(r.glyphs)
|
|
text = r.text or ""
|
|
for i, g in enumerate(glyphs):
|
|
ch = g.text if getattr(g, "text", None) is not None else (text[i] if i < len(text) else "?")
|
|
if i + 1 < len(glyphs):
|
|
adv = glyphs[i + 1].origin_x - g.origin_x
|
|
else:
|
|
adv = g.bbox_w if g.bbox_w > 0 else (r.font_size or 12) * 0.5
|
|
rows.append({
|
|
"char": ch,
|
|
"origin_x": g.origin_x,
|
|
"origin_y": g.origin_y,
|
|
"advance": adv,
|
|
"bbox_x": g.bbox_x,
|
|
"bbox_y": g.bbox_y,
|
|
"bbox_w": g.bbox_w,
|
|
"bbox_h": g.bbox_h,
|
|
"font_size": g.font_size,
|
|
"font_name": g.font_name,
|
|
"fid": r.internal_font_id,
|
|
})
|
|
joined = "".join(g["char"] for g in rows)
|
|
# Prefer longest match starting at TARGET / TYPED
|
|
for needle in (want_prefix, TARGET):
|
|
start = joined.find(needle)
|
|
if start >= 0:
|
|
return rows[start : start + len(want_prefix)] if want_prefix.startswith(TARGET) else rows[start:start + len(needle)]
|
|
return rows
|
|
|
|
|
|
def decompress_streams(pdf_bytes: bytes) -> list[str]:
|
|
out = []
|
|
for m in re.finditer(rb"stream\r?\n(.*?)\r?\nendstream", pdf_bytes, re.DOTALL):
|
|
try:
|
|
out.append(zlib.decompress(m.group(1)).decode("latin-1", "replace"))
|
|
except Exception:
|
|
pass
|
|
return out
|
|
|
|
|
|
def parse_heading_emits(pdf_bytes: bytes, y_approx: float = 597.45) -> list[dict]:
|
|
"""Parse Identity-H / FXF* per-char emits near heading baseline."""
|
|
rows = []
|
|
for s in decompress_streams(pdf_bytes):
|
|
if "597.45" not in s and f"{y_approx}" not in s:
|
|
continue
|
|
for m in re.finditer(
|
|
r"1 0 0 1 ([0-9.+\-]+) ([0-9.+\-]+) Tm\s+/(FXF\d+) ([0-9.]+) Tf.*?\[<([0-9A-Fa-f]+)>\]\s*TJ",
|
|
s,
|
|
re.DOTALL,
|
|
):
|
|
y = float(m.group(2))
|
|
if abs(y - y_approx) > 0.5:
|
|
continue
|
|
rows.append({
|
|
"x": float(m.group(1)),
|
|
"y": y,
|
|
"font_res": m.group(3),
|
|
"font_size": float(m.group(4)),
|
|
"gid": int(m.group(5), 16),
|
|
})
|
|
rows.sort(key=lambda r: r["x"])
|
|
return rows
|
|
|
|
|
|
def find_font_for_res(pdf_bytes: bytes, res_name: str) -> dict:
|
|
"""Resolve /FXFn -> font obj BaseFont + FontFile2 sha if possible."""
|
|
# Find resource mapping then font dict — crude but enough for forensic
|
|
objs = {}
|
|
parts = re.split(rb"(\d+)\s+0\s+obj", pdf_bytes)
|
|
i = 1
|
|
while i + 1 < len(parts):
|
|
objs[int(parts[i].decode())] = parts[i + 1].split(b"endobj", 1)[0]
|
|
i += 2
|
|
font_obj = None
|
|
for body in objs.values():
|
|
t = body.decode("latin-1", "replace")
|
|
m = re.search(rf"/{res_name}\s+(\d+)\s+0\s+R", t)
|
|
if m:
|
|
font_obj = int(m.group(1))
|
|
break
|
|
if font_obj is None:
|
|
return {"res": res_name, "error": "unmapped"}
|
|
body = objs.get(font_obj, b"").decode("latin-1", "replace")
|
|
bf = re.search(r"/BaseFont\s*/([^\s/>\[]+)", body)
|
|
info = {"res": res_name, "font_obj": font_obj, "baseFont": bf.group(1) if bf else None}
|
|
# Descendant / FontDescriptor / FontFile2
|
|
dm = re.search(r"/DescendantFonts\s*\[\s*(\d+)\s+0\s+R", body)
|
|
target = font_obj
|
|
if dm:
|
|
target = int(dm.group(1))
|
|
body = objs.get(target, b"").decode("latin-1", "replace")
|
|
fd = re.search(r"/FontDescriptor\s+(\d+)\s+0\s+R", body)
|
|
if fd:
|
|
fdb = objs.get(int(fd.group(1)), b"").decode("latin-1", "replace")
|
|
ff = re.search(r"/FontFile2\s+(\d+)\s+0\s+R", fdb)
|
|
if ff:
|
|
ffb = objs.get(int(ff.group(1)), b"")
|
|
m = re.search(rb"stream\r?\n(.*?)\r?\nendstream", ffb, re.DOTALL)
|
|
if m:
|
|
raw = m.group(1)
|
|
try:
|
|
raw = zlib.decompress(raw)
|
|
except Exception:
|
|
pass
|
|
info["fontfile2_len"] = len(raw)
|
|
info["fontfile2_sha"] = sha(raw)
|
|
return info
|
|
|
|
|
|
def crop_glyph(doc, box, dpi=DPI, pad=1.5) -> tuple[bytes, int, int]:
|
|
page = doc.get_page(0)
|
|
ph = page.height
|
|
x0 = box["bbox_x"] - pad
|
|
y0 = box["bbox_y"] - pad
|
|
x1 = box["bbox_x"] + box["bbox_w"] + pad
|
|
y1 = box["bbox_y"] + box["bbox_h"] + pad
|
|
y_top = ph - y1
|
|
height = y1 - y0
|
|
w, h, raw = page.render_region_raw(dpi, y_top, height)
|
|
scale = dpi / 72.0
|
|
left = max(0, int(x0 * scale))
|
|
right = min(w, int(x1 * scale) + 1)
|
|
cw = max(1, right - left)
|
|
crop = bytearray(cw * h * 4)
|
|
for row in range(h):
|
|
src = (row * w + left) * 4
|
|
dst = row * cw * 4
|
|
crop[dst : dst + cw * 4] = raw[src : src + cw * 4]
|
|
return bytes(crop), cw, h
|
|
|
|
|
|
def outline_hash(rgba: bytes) -> str:
|
|
return sha(rgba)
|
|
|
|
|
|
def apply_reflow(pdf_path: Path, data: dict, label: str) -> bytes:
|
|
op = {"version": "1.0", "operations": [{
|
|
"id": label, "type": "reflow_paragraph", "pageIndex": 0, "data": data,
|
|
}]}
|
|
doc = pdfengine.PdfDocument.load_from_file(str(pdf_path), "")
|
|
doc.apply_edits(json.dumps(op))
|
|
return doc.save_full()
|
|
|
|
|
|
def main():
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
print("=== LOAD ORIGINAL ===")
|
|
doc0 = pdfengine.PdfDocument.load_from_file(str(PDF), "")
|
|
_, para, text = find_para(doc0)
|
|
print(f"para text={text!r}")
|
|
layout = compute_layout(para)
|
|
flat = extract_flat_runs(layout["seedRuns"], FID, layout["seedRuns"][0]["size"], layout["seedRuns"][0]["color"])
|
|
|
|
# --- BEFORE: edit-entry unchanged ---
|
|
data_entry = build_reflow_data(layout, flat, layout["origLines"], "key-before")
|
|
print("\n=== BEFORE (unchanged, with origLines+advances) ===")
|
|
before_bytes = apply_reflow(PDF, data_entry, "before")
|
|
(OUT / "keystroke_before.pdf").write_bytes(before_bytes)
|
|
|
|
# --- AFTER: first keystroke (+x), mirror editedRef with seed-advance merge ---
|
|
seed_text = TARGET
|
|
seed_adv = None
|
|
for r in flat:
|
|
if r.get("text") == TARGET and r.get("advances") and len(r["advances"]) == len(TARGET):
|
|
seed_adv = list(r["advances"])
|
|
break
|
|
typed_runs = []
|
|
for r in flat:
|
|
nr = {k: v for k, v in r.items()}
|
|
if nr.get("text") == TARGET:
|
|
nr["text"] = TYPED
|
|
if seed_adv is not None:
|
|
nr["advances"] = seed_adv
|
|
nr["advanceSeedText"] = seed_text
|
|
else:
|
|
nr.pop("advances", None)
|
|
typed_runs.append(nr)
|
|
if not any(r.get("text") == TYPED for r in typed_runs):
|
|
base = flat[0] if flat else {"internalFontId": FID, "fontSize": 12, "color": "#1a5276"}
|
|
typed_runs = [{
|
|
"text": TYPED,
|
|
"internalFontId": base.get("internalFontId") or FID,
|
|
"fontSize": base.get("fontSize") or 12,
|
|
"color": base.get("color") or "#1a5276",
|
|
**({"advances": seed_adv, "advanceSeedText": seed_text} if seed_adv else {}),
|
|
}]
|
|
|
|
data_after = build_reflow_data(layout, typed_runs, None, "key-after")
|
|
data_after.pop("lines", None)
|
|
print("\n=== AFTER (typed +x, advanceSeedText merge, no origLines) ===")
|
|
print(f" runs={[ (r.get('text'), len(r.get('advances') or []), r.get('advanceSeedText')) for r in typed_runs ]}")
|
|
after_bytes = apply_reflow(PDF, data_after, "after")
|
|
(OUT / "keystroke_after.pdf").write_bytes(after_bytes)
|
|
|
|
# Load both for extraction + crops
|
|
doc_b = pdfengine.PdfDocument.load_from_memory(before_bytes, "")
|
|
doc_a = pdfengine.PdfDocument.load_from_memory(after_bytes, "")
|
|
_, para_b, text_b = find_para(doc_b)
|
|
_, para_a, text_a = find_para(doc_a)
|
|
print(f"before extracted={text_b!r}")
|
|
print(f"after extracted={text_a!r}")
|
|
|
|
glyphs_b = collect_glyphs(para_b, TARGET)
|
|
glyphs_a = collect_glyphs(para_a, TYPED)
|
|
print(f"before glyphs={len(glyphs_b)} after glyphs={len(glyphs_a)}")
|
|
|
|
emits_b = parse_heading_emits(before_bytes)
|
|
emits_a = parse_heading_emits(after_bytes)
|
|
print(f"before stream emits={len(emits_b)} after stream emits={len(emits_a)}")
|
|
|
|
fonts_b = {}
|
|
fonts_a = {}
|
|
for e in emits_b:
|
|
fonts_b.setdefault(e["font_res"], find_font_for_res(before_bytes, e["font_res"]))
|
|
for e in emits_a:
|
|
fonts_a.setdefault(e["font_res"], find_font_for_res(after_bytes, e["font_res"]))
|
|
print("before fonts:", json.dumps(fonts_b, indent=2))
|
|
print("after fonts:", json.dumps(fonts_a, indent=2))
|
|
|
|
# Align by index over shared prefix TARGET (ignore trailing x for pairwise compare of originals)
|
|
n = min(len(glyphs_b), len(glyphs_a), len(TARGET))
|
|
# Map stream emits to non-space chars
|
|
def attach_emits(glyphs, emits):
|
|
ei = 0
|
|
for g in glyphs:
|
|
if g["char"] == " ":
|
|
g["emitted"] = False
|
|
g["gid"] = None
|
|
g["font_res"] = None
|
|
g["stream_x"] = None
|
|
continue
|
|
if ei < len(emits):
|
|
e = emits[ei]
|
|
ei += 1
|
|
g["emitted"] = True
|
|
g["gid"] = e["gid"]
|
|
g["font_res"] = e["font_res"]
|
|
g["stream_x"] = e["x"]
|
|
g["stream_y"] = e["y"]
|
|
else:
|
|
g["emitted"] = False
|
|
g["gid"] = None
|
|
g["font_res"] = None
|
|
|
|
attach_emits(glyphs_b, emits_b)
|
|
attach_emits(glyphs_a, emits_a)
|
|
|
|
print("\n=== PER-GLYPH BEFORE vs AFTER (shared prefix) ===")
|
|
print(
|
|
f"{'#':>2} {'ch':>3} {'gidB':>5} {'gidA':>5} "
|
|
f"{'xB':>10} {'xA':>10} {'dx':>8} "
|
|
f"{'advB':>8} {'advA':>8} {'dAdv':>8} "
|
|
f"{'fontB':>6} {'fontA':>6} {'outEq':>5}"
|
|
)
|
|
first_diff = None
|
|
rows = []
|
|
for i in range(n):
|
|
b, a = glyphs_b[i], glyphs_a[i]
|
|
xB = b.get("stream_x", b["origin_x"])
|
|
xA = a.get("stream_x", a["origin_x"])
|
|
if xB is None:
|
|
xB = b["origin_x"]
|
|
if xA is None:
|
|
xA = a["origin_x"]
|
|
dx = xA - xB
|
|
dadv = a["advance"] - b["advance"]
|
|
|
|
# Outline hash using BEFORE bbox for both (normalized) when possible
|
|
out_eq = None
|
|
hb = ha = None
|
|
try:
|
|
if b["char"] != " " and b.get("bbox_w", 0) > 0:
|
|
rb, wb, hb_ = crop_glyph(doc_b, b)
|
|
# Use same crop box on after doc
|
|
ra, wa, ha_ = crop_glyph(doc_a, b)
|
|
h = min(hb_, ha_)
|
|
w = min(wb, wa)
|
|
|
|
def trim(rgba, W, H, tw, th):
|
|
out = bytearray(tw * th * 4)
|
|
for y in range(th):
|
|
out[y * tw * 4:(y + 1) * tw * 4] = rgba[y * W * 4:y * W * 4 + tw * 4]
|
|
return bytes(out)
|
|
|
|
tb, ta = trim(rb, wb, hb_, w, h), trim(ra, wa, ha_, w, h)
|
|
hb, ha = outline_hash(tb), outline_hash(ta)
|
|
out_eq = tb == ta
|
|
except Exception as e:
|
|
out_eq = f"err:{e}"
|
|
|
|
font_changed = (b.get("font_res") != a.get("font_res")) or (
|
|
fonts_b.get(b.get("font_res") or "", {}).get("fontfile2_sha")
|
|
!= fonts_a.get(a.get("font_res") or "", {}).get("fontfile2_sha")
|
|
)
|
|
pos_diff = abs(dx) > TOL
|
|
adv_diff = abs(dadv) > TOL
|
|
gid_diff = b.get("gid") != a.get("gid")
|
|
outline_diff = out_eq is False
|
|
changed = pos_diff or adv_diff or gid_diff or outline_diff or font_changed or (b["char"] != a["char"])
|
|
|
|
if first_diff is None and changed:
|
|
first_diff = {
|
|
"index": i,
|
|
"char": b["char"],
|
|
"reasons": [r for r, c in [
|
|
("position", pos_diff), ("advance", adv_diff), ("gid", gid_diff),
|
|
("outline", outline_diff), ("font", font_changed), ("char", b["char"] != a["char"]),
|
|
] if c],
|
|
"xB": xB, "xA": xA, "dx": dx,
|
|
"advB": b["advance"], "advA": a["advance"], "dAdv": dadv,
|
|
"gidB": b.get("gid"), "gidA": a.get("gid"),
|
|
"fontB": b.get("font_res"), "fontA": a.get("font_res"),
|
|
"fontInfoB": fonts_b.get(b.get("font_res") or ""),
|
|
"fontInfoA": fonts_a.get(a.get("font_res") or ""),
|
|
"outlineHashB": hb, "outlineHashA": ha,
|
|
}
|
|
|
|
print(
|
|
f"{i:2d} {b['char']:>3} {str(b.get('gid')):>5} {str(a.get('gid')):>5} "
|
|
f"{xB:10.4f} {xA:10.4f} {dx:8.4f} "
|
|
f"{b['advance']:8.4f} {a['advance']:8.4f} {dadv:8.4f} "
|
|
f"{str(b.get('font_res')):>6} {str(a.get('font_res')):>6} {str(out_eq):>5}"
|
|
)
|
|
rows.append({
|
|
"i": i, "char": b["char"],
|
|
"gidB": b.get("gid"), "gidA": a.get("gid"),
|
|
"xB": xB, "xA": xA, "dx": dx,
|
|
"yB": b.get("stream_y", b["origin_y"]), "yA": a.get("stream_y", a["origin_y"]),
|
|
"advB": b["advance"], "advA": a["advance"], "dAdv": dadv,
|
|
"fontB": b.get("font_res"), "fontA": a.get("font_res"),
|
|
"outline_equal": out_eq,
|
|
"outlineHashB": hb, "outlineHashA": ha,
|
|
"changed": changed,
|
|
})
|
|
|
|
# Trailing typed char
|
|
if len(glyphs_a) > n:
|
|
g = glyphs_a[n]
|
|
print(f"\n+++ typed glyph[{n}] char={g['char']!r} x={g.get('stream_x', g['origin_x'])} "
|
|
f"adv={g['advance']:.4f} gid={g.get('gid')} font={g.get('font_res')}")
|
|
|
|
print("\n=== FIRST GLYPH THAT CHANGES (before → after keystroke) ===")
|
|
print(json.dumps(first_diff, indent=2))
|
|
|
|
report = {
|
|
"before_text": TARGET,
|
|
"after_text": TYPED,
|
|
"before_fonts": fonts_b,
|
|
"after_fonts": fonts_a,
|
|
"first_diff": first_diff,
|
|
"glyphs": rows,
|
|
"after_extra": glyphs_a[n:] if len(glyphs_a) > n else [],
|
|
}
|
|
(OUT / "keystroke_report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
|
|
print(f"\nWrote {OUT / 'keystroke_report.json'}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|