v2: richtungsgetrennte Schluessel, Offset im Schluessel, PWA

- Neues Schluesselmodell: pro Beziehung zwei Stroeme (send.key/recv.key),
  Sende-Offset lebt im Schluessel-Header und wird beim Senden fortgeschrieben.
  Loest two-time-pad bei bidirektionaler Nutzung deterministisch.
- Stream-Identifier im Bild -> falscher Schluessel wird erkannt statt
  Kauderwelsch zu liefern. Rollen-Guards (send/recv) gegen Fehlbenutzung.
- CLI genkey erzeugt gerichtetes Keypaar (meine_seite/partner_seite).
- Web-App auf v2 umgestellt; Senden liefert aktualisierten send.key zum
  Download (Browser-Sandbox kann nicht auf den Stick schreiben).
- PWA: manifest + service worker (offline) + icons.
- Interop CLI<->Web in beide Richtungen per Node-Test abgesichert.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-14 21:46:06 +00:00
parent ad0b309acf
commit 30127db247
9 changed files with 549 additions and 469 deletions

View File

@@ -6,19 +6,31 @@ Verschlüsselt Text mit einem echten One-Time-Pad (XOR) und versteckt den
Ciphertext in den niederwertigsten Bits eines Trägerfotos. Das Ergebnis sieht
aus wie ein ganz normales Bild, trägt aber die Nachricht bit-genau in sich.
WICHTIG:
* Verschickt das Stego-Bild IMMER als Datei/Dokument, niemals als "Foto"
Messenger jagen Fotos durch verlustbehaftete Kompression und zerstören dabei
die versteckten Bits. Als Dokument bleibt das PNG unangetastet.
* Das Pad (der Schlüssel) muss auf beiden Seiten byte-identisch sein und darf
NIE zweimal für dieselben Bytes benutzt werden. Der Offset-Zähler
(pad.key.offset) sorgt automatisch dafür, dass kein Byte doppelt verbraucht
wird — solange ihr ihn nicht manuell zurückdreht.
RICHTUNGSGETRENNTE SCHLÜSSEL
----------------------------
Jede Beziehung nutzt ZWEI unabhängige Schlüsselströme — einen pro Senderichtung.
Dadurch können sich zwei gleichzeitig sendende Parteien nie überschneiden
(kein „two-time-pad"). Jede Seite hält:
send.key — dein Sendeschlüssel. Trägt den Sende-Offset IM Header; er wandert
bei jeder Nachricht vor. Der Partner hat dieselben Bytes als recv.key.
recv.key — dein Empfangsschlüssel. Wird nur GELESEN (Offset kommt aus dem Bild),
nie verändert.
Der Sende-Offset lebt also im Schlüssel selbst. Legt man den Schlüssel auf einen
USB-Stick und hängt ihn nur ans gerade benutzte Gerät, gibt es physisch nur einen
einzigen Offset-Zähler — kein Auseinanderdriften über mehrere Geräte möglich.
WICHTIG
-------
* Stego-Bild IMMER als Datei/Dokument verschicken, nie als „Foto" — Messenger
komprimieren Fotos verlustbehaftet und zerstören die versteckten Bits.
* send.key niemals doppelt/parallel auf zwei Geräten bespielen (siehe oben).
Subcommands:
genkey Erzeugt ein echtes Zufalls-Pad.
hide Verschlüsselt + versteckt eine Nachricht in einem Trägerbild.
reveal Extrahiert + entschlüsselt eine Nachricht aus einem Stego-Bild.
genkey Erzeugt ein gerichtetes Schlüsselpaar für eine Beziehung.
hide Verschlüsselt + versteckt eine Nachricht (braucht send.key).
reveal Extrahiert + entschlüsselt eine Nachricht (braucht recv.key).
Abhängigkeit: Pillow (pip install Pillow)
"""
@@ -30,67 +42,92 @@ import sys
from PIL import Image
MAGIC = b"SPST" # Schattenpost-Container-Kennung
VERSION = 1
# Header: magic(4) + version(1) + offset(8, uint64 BE) + length(4, uint32 BE)
HEADER_FMT = ">4sBQI"
HEADER_LEN = struct.calcsize(HEADER_FMT)
# --- Schlüsseldatei ---------------------------------------------------------
KEY_MAGIC = b"SPK1"
KEY_VERSION = 1
ROLE_RECV = 0
ROLE_SEND = 1
# magic(4) ver(1) role(1) stream_id(8) send_offset(uint64 BE, 8) + 2 pad = 24
KEY_FMT = ">4sBB8sQ2x"
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)
# --------------------------------------------------------------------------- #
# One-Time-Pad
# --------------------------------------------------------------------------- #
def otp_xor(data: bytes, key: bytes) -> bytes:
"""XOR data gegen key. key muss mindestens so lang wie data sein."""
if len(key) < len(data):
def otp_xor(data: bytes, material: bytes, offset: int) -> bytes:
if offset + len(data) > len(material):
raise ValueError(
f"Pad zu kurz: brauche {len(data)} Bytes, habe nur {len(key)}."
f"Schlüssel erschöpft: ab Offset {offset} brauche ich {len(data)} "
f"Bytes, habe nur {len(material) - offset}."
)
return bytes(d ^ k for d, k in zip(data, key))
return bytes(d ^ material[offset + i] for i, d in enumerate(data))
def load_pad(path: str) -> bytes:
# --------------------------------------------------------------------------- #
# Schlüsseldateien
# --------------------------------------------------------------------------- #
def read_key(path: str) -> dict:
with open(path, "rb") as fh:
return fh.read()
raw = fh.read()
if len(raw) < KEY_HEADER_LEN:
raise ValueError(f"{path}: keine gültige Schlüsseldatei (zu kurz).")
magic, ver, role, stream_id, send_offset = struct.unpack(
KEY_FMT, raw[:KEY_HEADER_LEN]
)
if magic != KEY_MAGIC:
raise ValueError(f"{path}: keine schattenpost-Schlüsseldatei.")
if ver != KEY_VERSION:
raise ValueError(f"{path}: unbekannte Schlüsselversion {ver}.")
return {
"role": role,
"stream_id": stream_id,
"send_offset": send_offset,
"material": raw[KEY_HEADER_LEN:],
}
def read_offset(pad_path: str) -> int:
"""Nächster unbenutzter Offset im Pad. Fehlt die Datei -> 0."""
off_path = pad_path + ".offset"
if not os.path.exists(off_path):
return 0
with open(off_path, "r") as fh:
return int(fh.read().strip() or "0")
def write_key(path: str, role: int, stream_id: bytes, send_offset: int,
material: bytes) -> None:
header = struct.pack(KEY_FMT, KEY_MAGIC, KEY_VERSION, role, stream_id,
send_offset)
with open(path, "wb") as fh:
fh.write(header + material)
def write_offset(pad_path: str, value: int) -> None:
with open(pad_path + ".offset", "w") as fh:
fh.write(str(value))
def advance_send_offset(path: str, new_offset: int) -> None:
"""Schreibt nur das 8-Byte-Offset-Feld in-place (kein Rewrite des Materials)."""
with open(path, "r+b") as fh:
fh.seek(KEY_OFFSET_POS)
fh.write(struct.pack(">Q", new_offset))
# --------------------------------------------------------------------------- #
# LSB-Steganografie
# --------------------------------------------------------------------------- #
def _bits_from_bytes(data: bytes):
"""Liefert die Bits von data, MSB zuerst pro Byte."""
for byte in data:
for i in range(7, -1, -1):
yield (byte >> i) & 1
def embed(carrier_path: str, payload: bytes, out_path: str) -> None:
"""Bettet payload in die LSBs von carrier ein und speichert als PNG."""
img = Image.open(carrier_path).convert("RGB")
width, height = img.size
capacity_bits = width * height * 3
need_bits = len(payload) * 8
if need_bits > capacity_bits:
if need_bits > width * height * 3:
raise ValueError(
f"Trägerbild zu klein: brauche {need_bits} Bits, "
f"Platz für {capacity_bits} ({need_bits // 8} vs "
f"{capacity_bits // 8} Bytes). Nimm ein größeres Foto."
f"Trägerbild zu klein: brauche {need_bits // 8} Bytes, Platz für "
f"{width * height * 3 // 8}. Nimm ein größeres Foto."
)
px = img.load()
bits = _bits_from_bytes(payload)
done = False
@@ -99,62 +136,55 @@ def embed(carrier_path: str, payload: bytes, out_path: str) -> None:
break
for x in range(width):
r, g, b = px[x, y]
channels = [r, g, b]
ch = [r, g, b]
for c in range(3):
bit = next(bits, None)
if bit is None:
done = True
break
channels[c] = (channels[c] & ~1) | bit
px[x, y] = (channels[0], channels[1], channels[2])
ch[c] = (ch[c] & ~1) | bit
px[x, y] = (ch[0], ch[1], ch[2])
if done:
break
img.save(out_path, "PNG")
def _read_bits(px, width, height, n_bits):
"""Liest n_bits LSBs linear aus dem Bild und gibt sie als Bytes zurück."""
out = bytearray()
acc = 0
count = 0
read = 0
acc = count = read = 0
for y in range(height):
for x in range(width):
r, g, b = px[x, y]
for c in (r, g, b):
for c in px[x, y]:
acc = (acc << 1) | (c & 1)
count += 1
read += 1
if count == 8:
out.append(acc)
acc = 0
count = 0
acc = count = 0
if read >= n_bits:
return bytes(out)
return bytes(out)
def extract(stego_path: str) -> tuple[int, bytes]:
"""Liest Header + Ciphertext aus einem Stego-Bild. Gibt (offset, ct)."""
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, HEADER_LEN * 8)
magic, version, offset, length = struct.unpack(HEADER_FMT, header)
if magic != MAGIC:
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 vielleicht doch als Foto komprimiert?"
"Kein schattenpost-Container gefunden (Magic passt nicht). "
"Wurde das Bild als Foto verlustbehaftet komprimiert?"
)
if version != VERSION:
raise ValueError(f"Unbekannte Version {version}.")
total_bits = (HEADER_LEN + length) * 8
blob = _read_bits(px, width, height, total_bits)
ciphertext = blob[HEADER_LEN:HEADER_LEN + length]
return offset, ciphertext
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],
}
# --------------------------------------------------------------------------- #
@@ -162,15 +192,29 @@ def extract(stego_path: str) -> tuple[int, bytes]:
# --------------------------------------------------------------------------- #
def cmd_genkey(args):
size = args.size
if os.path.exists(args.output) and not args.force:
sys.exit(f"[!] {args.output} existiert schon. --force zum Überschreiben.")
with open(args.output, "wb") as fh:
fh.write(os.urandom(size))
write_offset(args.output, 0)
print(f"[+] Pad erzeugt: {args.output} ({size} Bytes, echtes os.urandom)")
print(f"[+] Offset-Zähler zurückgesetzt: {args.output}.offset -> 0")
print("[i] Gib dieses Pad sicher an deinen Gesprächspartner weiter "
"(z.B. USB-Stick, persönlich). Beide brauchen dieselbe Datei.")
mine = args.output + "_meine_seite"
theirs = args.output + "_partner_seite"
for d in (mine, theirs):
if os.path.exists(d) and not args.force:
sys.exit(f"[!] {d} existiert schon. --force zum Überschreiben.")
os.makedirs(d, exist_ok=True)
# Zwei unabhängige Ströme: sid_out = ich->Partner, sid_in = Partner->ich
sid_out, sid_in = os.urandom(8), os.urandom(8)
mat_out, mat_in = os.urandom(size), os.urandom(size)
# Meine Seite: sende mit Strom OUT, empfange Strom IN
write_key(os.path.join(mine, "send.key"), ROLE_SEND, sid_out, 0, mat_out)
write_key(os.path.join(mine, "recv.key"), ROLE_RECV, sid_in, 0, mat_in)
# Partnerseite (spiegelverkehrt): sendet mit Strom IN, empfängt Strom OUT
write_key(os.path.join(theirs, "send.key"), ROLE_SEND, sid_in, 0, mat_in)
write_key(os.path.join(theirs, "recv.key"), ROLE_RECV, sid_out, 0, mat_out)
print(f"[+] Schlüsselpaar erzeugt ({size} Bytes je Richtung, echter Zufall).")
print(f" {mine}/ <- DU behältst diesen Ordner")
print(f" {theirs}/ <- gib DIESEN Ordner dem Partner (Stick, persönlich)")
print("[i] Jede Seite hat send.key (zum Senden) + recv.key (zum Empfangen).")
print("[i] send.key nur an EINEM Gerät gleichzeitig benutzen.")
def cmd_hide(args):
@@ -181,49 +225,53 @@ def cmd_hide(args):
plaintext = fh.read()
else:
plaintext = sys.stdin.buffer.read()
if not plaintext:
sys.exit("[!] Leere Nachricht.")
pad = load_pad(args.key)
offset = read_offset(args.key)
segment = pad[offset:offset + len(plaintext)]
key = read_key(args.key)
if key["role"] != ROLE_SEND:
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):
sys.exit(
f"[!] Pad erschöpft: ab Offset {offset} bleiben nur "
f"{len(segment)} Bytes, Nachricht braucht {len(plaintext)}. "
"Erzeuge/tausche ein neues Pad."
)
sys.exit(f"[!] Sendeschlüssel erschöpft: ab Offset {offset} bleiben nur "
f"{len(segment)} Bytes, Nachricht braucht {len(plaintext)}. "
"Erzeugt ein neues Schlüsselpaar.")
ciphertext = otp_xor(plaintext, segment)
header = struct.pack(HEADER_FMT, MAGIC, VERSION, offset, len(ciphertext))
payload = header + ciphertext
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, args.output)
advance_send_offset(args.key, offset + len(plaintext))
embed(args.carrier, payload, args.output)
write_offset(args.key, offset + len(plaintext))
print(f"[+] Nachricht verschlüsselt ({len(plaintext)} Bytes) "
f"mit Pad-Offset {offset}.")
print(f"[+] Verschlüsselt ({len(plaintext)} B) ab Sende-Offset {offset}.")
print(f"[+] Versteckt in: {args.output}")
print(f"[+] Pad-Offset weitergezählt -> {offset + len(plaintext)}")
print("[i] Verschicke das PNG als DATEI/Dokument, nicht als Foto!")
print(f"[+] Sende-Offset im Schlüssel weitergezählt -> "
f"{offset + len(plaintext)}")
print("[i] Als DATEI/Dokument verschicken, nicht als Foto!")
def cmd_reveal(args):
offset, ciphertext = extract(args.stego)
pad = load_pad(args.key)
segment = pad[offset:offset + len(ciphertext)]
if len(segment) < len(ciphertext):
sys.exit(
f"[!] Pad passt nicht: ab Offset {offset} brauche ich "
f"{len(ciphertext)} Bytes, habe nur {len(segment)}."
)
plaintext = otp_xor(ciphertext, segment)
info = extract(args.stego)
key = read_key(args.key)
if key["role"] != ROLE_RECV:
sys.exit("[!] Das ist ein Sendeschlüssel (send.key). Zum Empfangen "
"brauchst du deinen recv.key.")
if info["stream_id"] != key["stream_id"]:
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)}.")
plaintext = otp_xor(ct, key["material"], offset)
if args.outfile:
with open(args.outfile, "wb") as fh:
fh.write(plaintext)
print(f"[+] Entschlüsselt ({len(plaintext)} Bytes, Offset {offset}) "
print(f"[+] Entschlüsselt ({len(plaintext)} B, Offset {offset}) "
f"-> {args.outfile}")
else:
try:
@@ -237,31 +285,31 @@ def cmd_reveal(args):
def build_parser():
p = argparse.ArgumentParser(
prog="schattenpost",
description="One-Time-Pad + LSB-Steganografie für vertrauliche Nachrichten.",
description="One-Time-Pad + LSB-Steganografie, richtungsgetrennte Schlüssel.",
)
sub = p.add_subparsers(dest="cmd", required=True)
g = sub.add_parser("genkey", help="Echtes Zufalls-Pad erzeugen.")
g.add_argument("-o", "--output", required=True, help="Pad-Datei.")
g = sub.add_parser("genkey", help="Gerichtetes Schlüsselpaar erzeugen.")
g.add_argument("-o", "--output", required=True,
help="Basisname (erzeugt <name>_meine_seite/ + _partner_seite/).")
g.add_argument("-s", "--size", type=int, default=1_000_000,
help="Größe in Bytes (Default 1 MB).")
g.add_argument("--force", action="store_true", help="Überschreiben erlauben.")
help="Bytes je Richtung (Default 1 MB).")
g.add_argument("--force", action="store_true")
g.set_defaults(func=cmd_genkey)
h = sub.add_parser("hide", help="Nachricht verschlüsseln + verstecken.")
h.add_argument("-t", "--text", help="Nachrichtentext (sonst --infile/stdin).")
h.add_argument("-i", "--infile", help="Datei als Nachricht.")
h.add_argument("-k", "--key", required=True, help="Pad-Datei.")
h.add_argument("-t", "--text")
h.add_argument("-i", "--infile")
h.add_argument("-k", "--key", required=True, help="Dein send.key.")
h.add_argument("-c", "--carrier", required=True, help="Trägerbild.")
h.add_argument("-o", "--output", required=True, help="Stego-PNG-Ausgabe.")
h.add_argument("-o", "--output", required=True, help="Stego-PNG.")
h.set_defaults(func=cmd_hide)
r = sub.add_parser("reveal", help="Nachricht extrahieren + entschlüsseln.")
r.add_argument("-s", "--stego", required=True, help="Stego-PNG.")
r.add_argument("-k", "--key", required=True, help="Pad-Datei.")
r.add_argument("-o", "--outfile", help="Ausgabedatei (sonst stdout).")
r.add_argument("-s", "--stego", required=True)
r.add_argument("-k", "--key", required=True, help="Dein recv.key.")
r.add_argument("-o", "--outfile")
r.set_defaults(func=cmd_reveal)
return p