spack.py

"""
Source Packer (spack): Pack/Unpack directories into a single .spk file.

Archiv mit reiner Textausgabe mit 64 erlaubten Zeichen:
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ+-"

Ablauf packen:
  - blob bauen (Manifest + Daten), mit zlib komprimieren
  - ggf. AES-GCM verschlüsseln
  - kompletten Archiv-Bytestrom (Header+Payload) in obiges 64er Alphabet kodieren
  - als Textdatei schreiben (nur erlaubte Zeichen, keine Padding-Zeichen)

Ablauf entpacken:
  - .spk als Text lesen, 64er Alphabet zurück in Bytes decodieren
  - wie bisher: Header parsen, ggf. entschlüsseln, ggf. dekomprimieren, Dateien schreiben

Format (binär vor der Textkodierung):
  [magic 'SPK1' (4B)] [version=1 (1B)] [flags (1B)]
  if ENCRYPTED:
      [alg: b'AESG' (4B)] [salt_len (1B)] [salt (salt_len B)]
      [nonce_len (1B)] [nonce (nonce_len B)]
  payload = zlib.compress( blob )
  blob = [manifest_len (4B, big-endian)] [manifest_json (UTF-8)] [data_concat]

Password:
  - If given, payload is encrypted with AES-256-GCM (cryptography).
  - Key = PBKDF2-HMAC-SHA256(pass, salt, 200_000, 32 bytes)
  - salt: random 16 bytes, nonce: random 12 bytes
"""

import argparse
import fnmatch
import json
import os
import struct
import sys
import time
from io import BytesIO
from pathlib import Path
import zlib
import secrets
from typing import Optional, Tuple, List, Dict, Any

# Optional strong crypto (recommended for password usage)
try:
    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
    from cryptography.hazmat.primitives import hashes
    _HAS_CRYPTO = True
except Exception:
    _HAS_CRYPTO = False


MAGIC = b"SPK1"
VERSION = 1
FLAG_ENCRYPTED = 0x01

# --- 64-Zeichen-Alphabet & Codec (unpadded Base64-artig) ---
# Map: value (0..63) -> char & umgekehrt
_B64_CHARS = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ+-"
_B64_REV: Dict[str, int] = {ch: i for i, ch in enumerate(_B64_CHARS)}

def _b64_64_encode(data: bytes) -> str:
    """Unpadded Base64-artige Kodierung mit eigenem Alphabet (nur 64 erlaubte Zeichen)."""
    out: List[str] = []
    n = len(data)
    i = 0
    while i + 3 <= n:
        b1, b2, b3 = data[i], data[i+1], data[i+2]
        i += 3
        v1 = (b1 >> 2) & 0x3F
        v2 = ((b1 & 0x03) << 4) | (b2 >> 4)
        v3 = ((b2 & 0x0F) << 2) | (b3 >> 6)
        v4 = b3 & 0x3F
        out.append(_B64_CHARS[v1])
        out.append(_B64_CHARS[v2])
        out.append(_B64_CHARS[v3])
        out.append(_B64_CHARS[v4])
    rem = n - i
    if rem == 1:
        b1 = data[i]
        v1 = (b1 >> 2) & 0x3F
        v2 = (b1 & 0x03) << 4
        out.append(_B64_CHARS[v1])
        out.append(_B64_CHARS[v2])
        # keine Padding-Zeichen
    elif rem == 2:
        b1, b2 = data[i], data[i+1]
        v1 = (b1 >> 2) & 0x3F
        v2 = ((b1 & 0x03) << 4) | (b2 >> 4)
        v3 = (b2 & 0x0F) << 2
        out.append(_B64_CHARS[v1])
        out.append(_B64_CHARS[v2])
        out.append(_B64_CHARS[v3])
    return "".join(out)

def _b64_64_decode(s: str) -> bytes:
    """Dekodiert String (ohne fremde Whitespaces/Padding) zurück zu Bytes."""
    # Sicherheitscheck: nur erlaubte Zeichen
    for ch in s:
        if ch not in _B64_REV:
            raise ValueError("Ungültiges Zeichen in .spk")
    out = bytearray()
    n = len(s)
    i = 0
    # Volle 4er-Blöcke
    while i + 4 <= n:
        v1 = _B64_REV[s[i]]
        v2 = _B64_REV[s[i+1]]
        v3 = _B64_REV[s[i+2]]
        v4 = _B64_REV[s[i+3]]
        i += 4
        b1 = (v1 << 2) | (v2 >> 4)
        b2 = ((v2 & 0x0F) << 4) | (v3 >> 2)
        b3 = ((v3 & 0x03) << 6) | v4
        out.extend([b1, b2, b3])
    # Rest 2 oder 3 Zeichen (unpadded)
    rem = n - i
    if rem == 2:
        v1 = _B64_REV[s[i]]
        v2 = _B64_REV[s[i+1]]
        b1 = (v1 << 2) | (v2 >> 4)
        out.append(b1)
    elif rem == 3:
        v1 = _B64_REV[s[i]]
        v2 = _B64_REV[s[i+1]]
        v3 = _B64_REV[s[i+2]]
        b1 = (v1 << 2) | (v2 >> 4)
        b2 = ((v2 & 0x0F) << 4) | (v3 >> 2)
        out.extend([b1, b2])
    elif rem == 1:
        # 1 Restzeichen ist im Base64-Sinn unzulässig
        raise ValueError("Ungültige Länge der kodierten Daten (Rest=1).")
    return bytes(out)
# --- Ende Codec ---


def read_spignore(root: Path) -> List[str]:
    """
    Parse .spignore file (if present) into glob patterns.
    Supports:
      - comments starting with '#'
      - blank lines
      - glob patterns (*, ?, **)
      - trailing '/' means directory prefix match
    Patterns are matched against POSIX-style relative paths (with '/').
    """
    patterns: List[str] = []
    path = root / ".spignore"
    if not path.is_file():
        return patterns
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        patterns.append(line)
    return patterns


def path_matches_any(rel_posix: str, patterns: List[str]) -> bool:
    """Return True if rel path matches any ignore pattern."""
    for pat in patterns:
        if pat.endswith("/"):
            dir_prefix = pat.rstrip("/")
            if rel_posix == dir_prefix or rel_posix.startswith(dir_prefix + "/"):
                return True
        if fnmatch.fnmatch(rel_posix, pat):
            return True
    return False


def collect_entries(root: Path, patterns: List[str]) -> List[Dict[str, Any]]:
    """
    Walk directory, filter by .spignore patterns, and collect metadata.
    Returns entries sorted for stable packing.
    """
    entries: List[Dict[str, Any]] = []
    for p in sorted(root.rglob("*")):
        if not p.is_file():
            continue
        rel = p.relative_to(root).as_posix()
        if path_matches_any(rel, patterns):
            continue
        st = p.stat()
        entries.append({
            "path": rel,
            "size": int(st.st_size),
            "mtime": float(st.st_mtime),
            "mode": int(st.st_mode & 0o777),  # store posix perms
        })
    return entries


def build_blob_and_manifest(root: Path, entries: List[Dict[str, Any]]) -> bytes:
    """
    Read file bytes in order, compute offsets, build manifest, and prepend it.
    Returns the uncompressed blob ready for zlib.
    """
    data = BytesIO()
    offset = 0
    manifest_entries: List[Dict[str, Any]] = []

    for e in entries:
        file_path = root / e["path"]
        with file_path.open("rb") as f:
            chunk = f.read()
        length = len(chunk)
        assert length == e["size"], f"Size changed while reading: {e['path']}"
        data.write(chunk)
        manifest_entries.append({
            "path": e["path"],
            "size": e["size"],
            "mtime": e["mtime"],
            "mode": e["mode"],
            "offset": offset,
            "length": length,
        })
        offset += length

    manifest = {
        "root": root.name,
        "created_utc": int(time.time()),
        "entries": manifest_entries
    }
    m_bytes = json.dumps(manifest, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    blob = struct.pack(">I", len(m_bytes)) + m_bytes + data.getvalue()
    return blob


def zlib_compress(data: bytes) -> bytes:
    return zlib.compress(data, level=9)


def zlib_decompress(data: bytes) -> bytes:
    return zlib.decompress(data)


def derive_key(password: str, salt: bytes) -> bytes:
    if not _HAS_CRYPTO:
        raise RuntimeError(
            "Passwortschutz benötigt das Paket 'cryptography'. Installiere mit: pip install cryptography"
        )
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        iterations=200_000,
    )
    return kdf.derive(password.encode("utf-8"))


def encrypt(payload: bytes, password: str) -> Tuple[bytes, bytes, bytes]:
    """
    Returns (ciphertext, salt, nonce)
    """
    salt = secrets.token_bytes(16)
    nonce = secrets.token_bytes(12)
    key = derive_key(password, salt)
    aesgcm = AESGCM(key)
    ct = aesgcm.encrypt(nonce, payload, associated_data=None)
    return ct, salt, nonce


def decrypt(ciphertext: bytes, password: str, salt: bytes, nonce: bytes) -> bytes:
    key = derive_key(password, salt)
    aesgcm = AESGCM(key)
    return aesgcm.decrypt(nonce, ciphertext, associated_data=None)


def write_archive(out_path: Path, payload_compressed: bytes, password: Optional[str]) -> None:
    """
    Build full binary archive (header + payload), then encode to 64-char alphabet and write as text.
    """
    flags = 0
    parts: List[bytes] = [MAGIC, bytes([VERSION])]

    if password:
        flags |= FLAG_ENCRYPTED
        parts.append(bytes([flags]))
        # Encrypt compressed payload
        ct, salt, nonce = encrypt(payload_compressed, password)
        parts.append(b"AESG")
        parts.append(bytes([len(salt)]))
        parts.append(salt)
        parts.append(bytes([len(nonce)]))
        parts.append(nonce)
        parts.append(ct)
    else:
        parts.append(bytes([flags]))
        parts.append(payload_compressed)

    # Alles zu einem Bytestrom zusammenfügen
    binary_archive = b"".join(parts)
    # In 64er Alphabet kodieren
    encoded_text = _b64_64_encode(binary_archive)

    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8", newline="") as f:
        f.write(encoded_text)


def read_archive(in_path: Path, password: Optional[str]) -> bytes:
    """
    Read .spk text file, decode to binary archive, then return (possibly encrypted) compressed payload (as bytes).
    """
    # Als Text lesen (nur erlaubte Zeichen werden akzeptiert)
    encoded_text = Path(in_path).read_text(encoding="utf-8")
    encoded_text = encoded_text.strip()

    # In Bytes zurückwandeln
    data = _b64_64_decode(encoded_text)
    view = memoryview(data)
    pos = 0

    if view[pos:pos+4].tobytes() != MAGIC:
        raise ValueError("Ungültige Datei: MAGIC mismatch.")
    pos += 4

    version = view[pos]
    pos += 1
    if version != VERSION:
        raise ValueError(f"Unsupported version: {version}")

    flags = view[pos]
    pos += 1
    encrypted = bool(flags & FLAG_ENCRYPTED)

    if encrypted:
        alg = view[pos:pos+4].tobytes(); pos += 4
        if alg != b"AESG":
            raise ValueError("Unbekannter Verschlüsselungsalgorithmus.")
        salt_len = view[pos]; pos += 1
        salt = view[pos:pos+salt_len].tobytes(); pos += salt_len
        nonce_len = view[pos]; pos += 1
        nonce = view[pos:pos+nonce_len].tobytes(); pos += nonce_len
        ciphertext = view[pos:].tobytes()

        if not password:
            raise RuntimeError("Diese Datei ist verschlüsselt. Passwort angeben mit --password.")
        try:
            payload = decrypt(ciphertext, password, salt, nonce)
        except Exception as e:
            raise RuntimeError("Entschlüsselung fehlgeschlagen (falsches Passwort?).") from e
    else:
        payload = view[pos:].tobytes()

    return payload


def do_pack(src_dir: Path, out_file: Path, password: Optional[str]) -> None:
    if not src_dir.is_dir():
        raise FileNotFoundError(f"Quellverzeichnis nicht gefunden: {src_dir}")
    patterns = read_spignore(src_dir)
    entries = collect_entries(src_dir, patterns)
    blob = build_blob_and_manifest(src_dir, entries)
    compressed = zlib_compress(blob)
    write_archive(out_file, compressed, password)
    print(f"OK: {len(entries)} Dateien gepackt → {out_file}")


def restore_permissions(path: Path, mode: int) -> None:
    try:
        path.chmod(mode)
    except Exception:
        # Not fatal on Windows; ignore silently
        pass


def do_unpack(in_file: Path, out_dir: Path, password: Optional[str]) -> None:
    if not in_file.is_file():
        raise FileNotFoundError(f"Archiv nicht gefunden: {in_file}")
    compressed = read_archive(in_file, password)
    blob = zlib_decompress(compressed)

    # Parse manifest
    if len(blob) < 4:
        raise ValueError("Beschädigtes Archiv (zu kurz).")
    (m_len,) = struct.unpack(">I", blob[:4])
    m_end = 4 + m_len
    m_bytes = blob[4:m_end]
    data_concat = blob[m_end:]
    manifest = json.loads(m_bytes.decode("utf-8"))

    entries = manifest.get("entries", [])
    for e in entries:
        rel = e["path"]
        target = out_dir / rel
        target.parent.mkdir(parents=True, exist_ok=True)
        start = e["offset"]
        end = start + e["length"]
        chunk = data_concat[start:end]
        with target.open("wb") as f:
            f.write(chunk)
        # mtime / mode
        try:
            os.utime(target, (time.time(), e.get("mtime", time.time())))
        except Exception:
            pass
        mode = e.get("mode")
        if mode is not None:
            restore_permissions(target, mode)

    print(f"OK: {len(entries)} Dateien entpackt → {out_dir}")


def build_argparser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="spack", description="Pack/Unpack Verzeichnisse in .spk-Archiv")
    sub = p.add_subparsers(dest="action", required=True)

    p_pack = sub.add_parser("pack", help="Verzeichnis packen")
    p_pack.add_argument("directory", type=Path, help="Quellverzeichnis")
    p_pack.add_argument("output", type=Path, help="Ausgabedatei (.spk - Text)")
    p_pack.add_argument("--password", "-p", type=str, default=None, help="Optionales Passwort (AES-GCM)")

    p_unpack = sub.add_parser("unpack", help="Archiv entpacken")
    p_unpack.add_argument("archive", type=Path, help="Archivdatei (.spk - Text)")
    p_unpack.add_argument("output_dir", type=Path, help="Zielverzeichnis")
    p_unpack.add_argument("--password", "-p", type=str, default=None, help="Passwort, falls verschlüsselt")

    return p


def main(argv=None):
    argv = argv if argv is not None else sys.argv[1:]
    args = build_argparser().parse_args(argv)

    try:
        if args.action == "pack":
            do_pack(args.directory, args.output, args.password)
        elif args.action == "unpack":
            do_unpack(args.archive, args.output_dir, args.password)
        else:
            raise ValueError("Unbekannte Aktion.")
    except Exception as e:
        print(f"Fehler: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()