SPS3: Poly1305-Einmal-MAC + Public-Ready README/LICENSE
Authentifizierung für CLI und Web: - Poly1305 (RFC 8439) pur in Python (bigint) und JS (BigInt), keine neue Dependency - MAC-Schluessel pro Nachricht frisch aus dem Pad (len+32), encrypt-then-MAC ueber Header + Ciphertext; reveal lehnt manipulierte/desynchrone Bilder hart ab - Stego-Format SPS2 -> SPS3 (16-B-Tag im Header); alte SPS2-Bilder werden noch gelesen, aber als unauthentifiziert markiert - Service-Worker-Cache auf v3 gebumpt - Interop-Runner (web/_interop_run.sh): RFC-Vektor cross-impl, beide Richtungen, Tamper-Erkennung README fuer Public ueberarbeitet (ehrliches Bedrohungsmodell, Voraussetzungen, Projektstruktur, Tests) + MIT-LICENSE. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Code
148
schattenpost.py
148
schattenpost.py
@@ -62,11 +62,19 @@ KEY_HEADER_LEN = struct.calcsize(KEY_FMT)
|
||||
KEY_OFFSET_POS = 14 # Byteposition des send_offset-Feldes (für In-Place-Update)
|
||||
|
||||
# --- Stego-Container (im Bild) ---------------------------------------------
|
||||
STEGO_MAGIC = b"SPS2"
|
||||
STEGO_VERSION = 2
|
||||
# magic(4) ver(1) stream_id(8) offset(uint64 BE, 8) length(uint32 BE, 4) = 25
|
||||
STEGO_FMT = ">4sB8sQI"
|
||||
STEGO_HEADER_LEN = struct.calcsize(STEGO_FMT)
|
||||
# v3 authentifiziert zusätzlich mit einem Poly1305-Einmal-MAC (Tag im Header).
|
||||
STEGO_MAGIC = b"SPS3"
|
||||
STEGO_VERSION = 3
|
||||
# Vor-Tag-Header: magic(4) ver(1) stream_id(8) offset(uint64 BE,8) len(uint32 BE,4)=25
|
||||
STEGO_PRE_FMT = ">4sB8sQI"
|
||||
STEGO_PRE_LEN = struct.calcsize(STEGO_PRE_FMT) # 25
|
||||
STEGO_TAG_LEN = 16 # Poly1305-Tag
|
||||
STEGO_HEADER_LEN = STEGO_PRE_LEN + STEGO_TAG_LEN # 41
|
||||
MAC_KEY_LEN = 32 # r(16)+s(16), aus dem Pad
|
||||
|
||||
# Alt-Format (nur noch beim Aufdecken akzeptiert, unauthentifiziert).
|
||||
STEGO_MAGIC_V2 = b"SPS2"
|
||||
STEGO_VERSION_V2 = 2
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -81,6 +89,40 @@ def otp_xor(data: bytes, material: bytes, offset: int) -> bytes:
|
||||
return bytes(d ^ material[offset + i] for i, d in enumerate(data))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Poly1305 — informationstheoretisch sicherer Einmal-MAC (RFC 8439)
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Der 32-Byte-Schlüssel (r|s) wird pro Nachricht frisch aus dem One-Time-Pad
|
||||
# gezogen (direkt hinter dem Nachrichten-Keystream) und NIE wiederverwendet —
|
||||
# genau die Bedingung, unter der Poly1305 fälschungssicher ist, ohne die
|
||||
# informationstheoretische Sicherheit des OTP aufzugeben.
|
||||
_P1305_P = (1 << 130) - 5
|
||||
_P1305_CLAMP = 0x0ffffffc0ffffffc0ffffffc0fffffff
|
||||
|
||||
|
||||
def poly1305_mac(msg: bytes, key: bytes) -> bytes:
|
||||
if len(key) != 32:
|
||||
raise ValueError("Poly1305-Schlüssel muss 32 Byte sein.")
|
||||
r = int.from_bytes(key[:16], "little") & _P1305_CLAMP
|
||||
s = int.from_bytes(key[16:32], "little")
|
||||
acc = 0
|
||||
for i in range(0, len(msg), 16):
|
||||
block = msg[i:i + 16]
|
||||
n = int.from_bytes(block, "little") | (1 << (8 * len(block)))
|
||||
acc = (acc + n) * r % _P1305_P
|
||||
return ((acc + s) % (1 << 128)).to_bytes(16, "little")
|
||||
|
||||
|
||||
def ct_equal(a: bytes, b: bytes) -> bool:
|
||||
"""Konstant-zeitiger Vergleich (kein Early-Exit-Timing-Leck)."""
|
||||
if len(a) != len(b):
|
||||
return False
|
||||
diff = 0
|
||||
for x, y in zip(a, b):
|
||||
diff |= x ^ y
|
||||
return diff == 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Schlüsseldateien
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -179,21 +221,36 @@ def extract(stego_path: str) -> dict:
|
||||
img = Image.open(stego_path).convert("RGB")
|
||||
width, height = img.size
|
||||
px = img.load()
|
||||
header = _read_bits(px, width, height, STEGO_HEADER_LEN * 8)
|
||||
magic, ver, stream_id, offset, length = struct.unpack(STEGO_FMT, header)
|
||||
if magic != STEGO_MAGIC:
|
||||
raise ValueError(
|
||||
"Kein schattenpost-Container gefunden (Magic passt nicht). "
|
||||
"Wurde das Bild als Foto verlustbehaftet komprimiert?"
|
||||
)
|
||||
if ver != STEGO_VERSION:
|
||||
raise ValueError(f"Unbekannte Container-Version {ver}.")
|
||||
blob = _read_bits(px, width, height, (STEGO_HEADER_LEN + length) * 8)
|
||||
return {
|
||||
"stream_id": stream_id,
|
||||
"offset": offset,
|
||||
"ciphertext": blob[STEGO_HEADER_LEN:STEGO_HEADER_LEN + length],
|
||||
}
|
||||
# Vor-Tag-Header ist bei v2 und v3 identisch aufgebaut (25 B) -> erst lesen.
|
||||
pre = _read_bits(px, width, height, STEGO_PRE_LEN * 8)
|
||||
magic, ver, stream_id, offset, length = struct.unpack(STEGO_PRE_FMT, pre)
|
||||
if magic == STEGO_MAGIC and ver == STEGO_VERSION:
|
||||
blob = _read_bits(px, width, height,
|
||||
(STEGO_HEADER_LEN + length) * 8)
|
||||
return {
|
||||
"version": ver,
|
||||
"stream_id": stream_id,
|
||||
"offset": offset,
|
||||
"length": length,
|
||||
"tag": blob[STEGO_PRE_LEN:STEGO_HEADER_LEN],
|
||||
"mac_input_header": blob[:STEGO_PRE_LEN],
|
||||
"ciphertext": blob[STEGO_HEADER_LEN:STEGO_HEADER_LEN + length],
|
||||
}
|
||||
if magic == STEGO_MAGIC_V2 and ver == STEGO_VERSION_V2:
|
||||
blob = _read_bits(px, width, height,
|
||||
(STEGO_PRE_LEN + length) * 8)
|
||||
return {
|
||||
"version": ver,
|
||||
"stream_id": stream_id,
|
||||
"offset": offset,
|
||||
"length": length,
|
||||
"tag": None,
|
||||
"ciphertext": blob[STEGO_PRE_LEN:STEGO_PRE_LEN + length],
|
||||
}
|
||||
raise ValueError(
|
||||
"Kein schattenpost-Container gefunden (Magic/Version passt nicht). "
|
||||
"Wurde das Bild als Foto verlustbehaftet komprimiert?"
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -243,22 +300,27 @@ def cmd_hide(args):
|
||||
sys.exit("[!] Das ist ein Empfangsschlüssel (recv.key). Zum Senden "
|
||||
"brauchst du deinen send.key.")
|
||||
offset = key["send_offset"]
|
||||
segment = key["material"][offset:offset + len(plaintext)]
|
||||
if len(segment) < len(plaintext):
|
||||
# Pro Nachricht: len(plaintext) Bytes Keystream + 32 Bytes MAC-Schlüssel.
|
||||
need = len(plaintext) + MAC_KEY_LEN
|
||||
segment = key["material"][offset:offset + need]
|
||||
if len(segment) < need:
|
||||
sys.exit(f"[!] Sendeschlüssel erschöpft: ab Offset {offset} bleiben nur "
|
||||
f"{len(segment)} Bytes, Nachricht braucht {len(plaintext)}. "
|
||||
f"{len(segment)} Bytes, Nachricht+MAC brauchen {need}. "
|
||||
"Erzeugt ein neues Schlüsselpaar.")
|
||||
|
||||
ciphertext = otp_xor(plaintext, key["material"], offset)
|
||||
header = struct.pack(STEGO_FMT, STEGO_MAGIC, STEGO_VERSION,
|
||||
key["stream_id"], offset, len(ciphertext))
|
||||
embed(args.carrier, header + ciphertext, out_path)
|
||||
advance_send_offset(args.key, offset + len(plaintext))
|
||||
mac_key = key["material"][offset + len(plaintext):offset + need]
|
||||
pre_header = struct.pack(STEGO_PRE_FMT, STEGO_MAGIC, STEGO_VERSION,
|
||||
key["stream_id"], offset, len(ciphertext))
|
||||
tag = poly1305_mac(pre_header + ciphertext, mac_key)
|
||||
embed(args.carrier, pre_header + tag + ciphertext, out_path)
|
||||
advance_send_offset(args.key, offset + need)
|
||||
|
||||
print(f"[+] Verschlüsselt ({len(plaintext)} B) ab Sende-Offset {offset}.")
|
||||
print(f"[+] Verschlüsselt + authentifiziert ({len(plaintext)} B) ab "
|
||||
f"Sende-Offset {offset}.")
|
||||
print(f"[+] Versteckt in: {out_path}")
|
||||
print(f"[+] Sende-Offset im Schlüssel weitergezählt -> "
|
||||
f"{offset + len(plaintext)}")
|
||||
print(f"[+] Sende-Offset im Schlüssel weitergezählt -> {offset + need} "
|
||||
f"(Nachricht {len(plaintext)} + MAC {MAC_KEY_LEN}).")
|
||||
print("[i] Als DATEI/Dokument verschicken, nicht als Foto!")
|
||||
|
||||
|
||||
@@ -272,10 +334,28 @@ def cmd_reveal(args):
|
||||
sys.exit("[!] Falscher Schlüssel: dieses Bild gehört zu einem anderen "
|
||||
"Kontakt/Strom. Nimm den passenden recv.key.")
|
||||
offset, ct = info["offset"], info["ciphertext"]
|
||||
segment = key["material"][offset:offset + len(ct)]
|
||||
if len(segment) < len(ct):
|
||||
sys.exit(f"[!] Schlüssel passt nicht: ab Offset {offset} brauche ich "
|
||||
f"{len(ct)} Bytes, habe nur {len(segment)}.")
|
||||
|
||||
if info["version"] == STEGO_VERSION:
|
||||
# Authentifiziert: MAC-Schlüssel liegt direkt hinter dem Keystream.
|
||||
need = len(ct) + MAC_KEY_LEN
|
||||
segment = key["material"][offset:offset + need]
|
||||
if len(segment) < need:
|
||||
sys.exit(f"[!] Schlüssel passt nicht: ab Offset {offset} brauche ich "
|
||||
f"{need} Bytes (Nachricht+MAC), habe nur {len(segment)}.")
|
||||
mac_key = key["material"][offset + len(ct):offset + need]
|
||||
expected = poly1305_mac(info["mac_input_header"] + ct, mac_key)
|
||||
if not ct_equal(expected, info["tag"]):
|
||||
sys.exit("[!] Authentifizierung fehlgeschlagen: das Bild wurde "
|
||||
"verändert, oder der Schlüssel ist nicht synchron "
|
||||
"(Offset-Drift). Nachricht NICHT vertrauen.")
|
||||
else:
|
||||
segment = key["material"][offset:offset + len(ct)]
|
||||
if len(segment) < len(ct):
|
||||
sys.exit(f"[!] Schlüssel passt nicht: ab Offset {offset} brauche ich "
|
||||
f"{len(ct)} Bytes, habe nur {len(segment)}.")
|
||||
print("[i] Warnung: altes SPS2-Bild — unauthentifiziert, "
|
||||
"keine Manipulationssicherheit.", file=sys.stderr)
|
||||
|
||||
plaintext = otp_xor(ct, key["material"], offset)
|
||||
|
||||
if args.outfile:
|
||||
|
||||
Reference in New Issue
Block a user