iglyph/iGlyph.py

#!/usr/bin/env python3
"""
iglyph - bundle .ttf/.otf files into a single iOS/iPadOS .mobileconfig

The Apple .mobileconfig spec has long supported multi-font payloads in
a single profile. iGlyph just... does that. Pure stdlib. The App Store
options (iFont, Fontastic) bundle too, but charge for it.

Usage:
    python3 iglyph.py ./my-fonts/
    python3 iglyph.py ./fonts/ -o brand.mobileconfig --name "lark.cx Fonts"
    python3 iglyph.py ./fonts/ --list
    python3 iglyph.py ./fonts/ --serve            # default port 18888
    python3 iglyph.py ./fonts/ --serve 8080 --timeout 30
"""

from __future__ import annotations

import argparse
import hashlib
import http.server
import plistlib
import re
import socket
import socketserver
import struct
import sys
import threading
import time
import uuid
from pathlib import Path

# =======================================================================
# Font validation and PostScript-name extraction (stdlib only)
# =======================================================================

# Magic bytes at offset 0 for supported font formats
_TTF_MAGIC = b"\x00\x01\x00\x00"
_OTF_MAGIC = b"OTTO"
_TRUE_MAGIC = b"true"  # legacy Apple TTF
_TTC_MAGIC = b"ttcf"  # font collection container (covers .ttc and .otc)
_WOFF_MAGIC = b"wOFF"
_WOFF2_MAGIC = b"wOF2"

_VALID_MAGICS = {_TTF_MAGIC, _OTF_MAGIC, _TRUE_MAGIC}


class FontError(ValueError):
    """Raised when a font file is invalid or unsupported"""


def validate_font_magic(path: Path) -> None:
    """Check the first 4 bytes look like a TTF or OTF

    Raises FontError with a helpful message for known-bad formats
    """
    if path.stat().st_size < 12:
        raise FontError(f"{path}: file too small to be a valid font")

    head = path.read_bytes()[:4]

    if head in _VALID_MAGICS:
        return
    if head == _TTC_MAGIC:
        raise FontError(
            f"{path}: this is a font collection (.ttc/.otc). "
            f"iOS doesn't support these; split it into individual .ttf/.otf "
            f"files with fontTools before bundling."
        )
    if head == _WOFF_MAGIC or head == _WOFF2_MAGIC:
        raise FontError(
            f"{path}: this is a web font (WOFF/WOFF2), not a TTF/OTF. "
            f"iOS won't install it. Convert it to TTF/OTF first."
        )
    raise FontError(
        f"{path}: doesn't look like a valid font "
        f"(magic bytes: {head!r}, expected TTF or OTF)"
    )


def extract_postscript_name(path: Path) -> str | None:
    """Parse the `name` table from a TTF/OTF and return the PostScript name

    Returns None if the name table can't be parsed or nameID 6 isn't present.
    Pure stdlib - no fontTools dependency.
    See: learn.microsoft.com/en-us/typography/opentype/spec/name
    """
    try:
        data = path.read_bytes()

        # sfnt header: u32 version, u16 numTables, u16 searchRange,
        # u16 entrySelector, u16 rangeShift
        if len(data) < 12:
            return None
        num_tables = struct.unpack(">H", data[4:6])[0]

        # Table directory: 16 bytes per entry - tag (4), checksum (4),
        # offset (4), length (4)
        name_offset = None
        name_length = None
        for i in range(num_tables):
            entry_start = 12 + i * 16
            if entry_start + 16 > len(data):
                return None
            tag = data[entry_start : entry_start + 4]
            if tag == b"name":
                name_offset = struct.unpack(
                    ">I", data[entry_start + 8 : entry_start + 12]
                )[0]
                name_length = struct.unpack(
                    ">I", data[entry_start + 12 : entry_start + 16]
                )[0]
                break

        if name_offset is None or name_length is None:
            return None
        if name_offset + name_length > len(data):
            return None

        # name table header: u16 format, u16 count, u16 stringOffset
        if name_length < 6:
            return None
        fmt, count, string_offset = struct.unpack(
            ">HHH", data[name_offset : name_offset + 6]
        )

        # Each name record is 12 bytes:
        # u16 platformID, u16 encodingID, u16 languageID, u16 nameID,
        # u16 length, u16 stringOffset (from start of storage area)
        record_size = 12
        records_start = name_offset + 6
        storage_start = name_offset + string_offset

        # Prefer Windows Unicode (3, 1), fall back to Macintosh Roman (1, 0)
        candidates: list[
            tuple[int, int, int, str]
        ] = []  # (priority, plat, enc, decoded)
        for i in range(count):
            rec_start = records_start + i * record_size
            if rec_start + record_size > len(data):
                break
            platform_id, encoding_id, _lang_id, name_id, length, offset = struct.unpack(
                ">HHHHHH", data[rec_start : rec_start + record_size]
            )
            if name_id != 6:  # 6 = PostScript name
                continue
            # Priority: Windows Unicode = 0 (best), Mac Roman = 1, others = 2
            if platform_id == 3 and encoding_id == 1:
                priority = 0
            elif platform_id == 1 and encoding_id == 0:
                priority = 1
            else:
                priority = 2
            str_start = storage_start + offset
            str_end = str_start + length
            if str_end > len(data):
                continue
            raw = data[str_start:str_end]
            try:
                if platform_id == 3:  # Windows: UTF-16BE
                    decoded = raw.decode("utf-16-be")
                elif platform_id == 1:  # Mac: MacRoman
                    decoded = raw.decode("mac-roman")
                else:
                    decoded = raw.decode("utf-8", errors="replace")
                candidates.append((priority, platform_id, encoding_id, decoded))
            except (UnicodeDecodeError, LookupError):
                continue

        if not candidates:
            return None
        candidates.sort(key=lambda c: c[0])
        return candidates[0][3]
    except (OSError, struct.error):
        return None


# =======================================================================
# Profile generation
# =======================================================================


def discover_fonts(inputs: list[Path]) -> list[Path]:
    """Recursively find .ttf and .otf files in the given inputs.

    Case-insensitive extension matching. Resolved paths used for dedup but
    user-visible paths preserved in the return (so error messages match
    what the user typed).
    """
    found: list[Path] = []
    for entry in inputs:
        if not entry.exists():
            raise FontError(f"input not found: {entry}")
        if entry.is_file():
            if entry.suffix.lower() in {".ttf", ".otf"}:
                found.append(entry)
            else:
                raise FontError(
                    f"{entry}: unsupported extension (only .ttf and .otf are accepted)"
                )
        elif entry.is_dir():
            for f in entry.rglob("*"):
                if f.is_file() and f.suffix.lower() in {".ttf", ".otf"}:
                    found.append(f)
    # Deduplicate by resolved path while preserving user-visible paths
    seen: set[Path] = set()
    unique: list[Path] = []
    for f in found:
        resolved = f.resolve()
        if resolved not in seen:
            seen.add(resolved)
            unique.append(f)
    return unique


def _slugify(text: str) -> str:
    """Lowercase, hyphenate, keep only [a-z0-9-]. For identifier derivation."""
    s = re.sub(r"[^a-zA-Z0-9]+", "-", text.lower()).strip("-")
    return s or "bundle"


def _build_description(
    annotated: list[tuple[Path, str | None]],
    max_listed: int = 40,
) -> str:
    """Build a description listing fonts

    Shows the PS name where available. Falls back to the filename stem.
    Truncates with a "+N more" tail if the list exceeds max_listed
    """
    count = len(annotated)
    names: list[str] = []
    for path, ps in annotated:
        names.append(ps if ps else path.stem)
    if count <= max_listed:
        listed = ", ".join(names)
    else:
        head = ", ".join(names[:max_listed])
        listed = f"{head}, +{count - max_listed} more"
    return f"Installs {count} font(s): {listed}"


def build_font_payload(font_path: Path, profile_uuid: str) -> dict:
    """Build a single com.apple.font payload dict for one font file"""
    payload_uuid = str(uuid.uuid4()).upper()
    return {
        "PayloadType": "com.apple.font",
        "PayloadVersion": 1,
        "PayloadIdentifier": f"{profile_uuid}.font.{payload_uuid}",
        "PayloadUUID": payload_uuid,
        "PayloadDisplayName": font_path.stem,
        "PayloadDescription": f"Installs {font_path.name}",
        "Name": font_path.name,
        "Font": font_path.read_bytes(),
    }


def build_profile(
    annotated: list[tuple[Path, str | None]],
    name: str,
    identifier: str,
    organization: str,
) -> bytes:
    """Build a complete .mobileconfig (XML plist) as bytes"""
    fonts = [f for f, _ in annotated]
    profile_uuid = str(uuid.uuid4()).upper()
    payloads = [build_font_payload(f, profile_uuid) for f in fonts]
    description = _build_description(annotated)
    profile = {
        "PayloadType": "Configuration",
        "PayloadVersion": 1,
        "PayloadIdentifier": identifier,
        "PayloadUUID": profile_uuid,
        "PayloadDisplayName": name,
        "PayloadDescription": description,
        "PayloadOrganization": organization,
        "PayloadScope": "User",
        "PayloadRemovalDisallowed": False,
        "PayloadContent": payloads,
    }
    return plistlib.dumps(profile, fmt=plistlib.FMT_XML, sort_keys=False)


def _file_sha256(path: Path) -> str:
    """Compute SHA-256 of a file's contents, chunked to handle large fonts."""
    h = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def derive_identifier(name: str, annotated: list[tuple[Path, str | None]]) -> str:
    """Derive a unique reverse-DNS identifier from the profile name + content.

    Hashes actual font bytes so that replacing a file with a different one
    that has the same PS name produces a new identifier (and thus coexists
    on the device rather than silently replacing). Sorted by PS name (or
    filename fallback) so identifier is independent of input ordering.
    """
    slug = _slugify(name)
    parts = [
        f"{ps or path.name}:{_file_sha256(path)}"
        for path, ps in sorted(annotated, key=lambda x: (x[1] or x[0].name).lower())
    ]
    content_hash = hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()[:8]
    return f"cx.lark.iglyph.{slug}.{content_hash}"


# =======================================================================
# Validation pipeline (magic bytes + PS-name collision detection)
# =======================================================================


def validate_and_report(
    fonts: list[Path],
) -> tuple[list[Path], list[tuple[Path, str | None]]]:
    """Validate all fonts, warn on PS-name collisions, return validated list.

    Returns (valid_fonts, [(path, ps_name), ...]).
    Raises FontError on hard validation failures (bad magic bytes, missing
    files). PS-name collisions are warned to stderr but not blocking;
    iOS silently drops duplicates. User can proceed if they want.
    """
    if not fonts:
        raise FontError("no .ttf or .otf files found in the input")

    annotated: list[tuple[Path, str | None]] = []
    for f in fonts:
        validate_font_magic(f)
        ps_name = extract_postscript_name(f)
        annotated.append((f, ps_name))

    # Detect PS-name collisions and warn (non-fatal)
    by_ps: dict[str, list[Path]] = {}
    for path, ps in annotated:
        if ps is None:
            continue
        by_ps.setdefault(ps, []).append(path)
    collisions = {ps: paths for ps, paths in by_ps.items() if len(paths) > 1}
    if collisions:
        lines = [
            "warning: PostScript-name collisions detected (iOS will silently keep one of each):"
        ]
        for ps, paths in collisions.items():
            lines.append(f"  {ps!r}:")
            for p in paths:
                lines.append(f"    - {p}")
        lines.append(
            "  proceeding anyway - remove duplicates if you want deterministic behavior"
        )
        print("\n".join(lines), file=sys.stderr)

    return [f for f, _ in annotated], annotated


# =======================================================================
# Network delivery
# =======================================================================


def get_local_ips() -> list[str]:
    """Enumerate all non-loopback IPv4 addresses for this host."""
    ips: list[str] = []
    try:
        hostname = socket.gethostname()
        for info in socket.getaddrinfo(hostname, None, socket.AF_INET):
            ip = info[4][0]
            if not ip.startswith("127.") and ip not in ips:
                ips.append(ip)
    except socket.gaierror:
        pass

    # Fallback: connect to a public IP and read back our local address
    if not ips:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
                s.connect(("8.8.8.8", 80))
                ip = s.getsockname()[0]
                if not ip.startswith("127."):
                    ips.append(ip)
        except OSError:
            pass

    return ips or ["127.0.0.1"]


class MobileConfigHandler(http.server.BaseHTTPRequestHandler):
    """Serves a single .mobileconfig file with the correct MIME type.

    Responds to any GET (or HEAD). Path doesn't matter. The Content-
    Disposition header carries the correct filename. User
    can hit just `http://192.168.1.47:18888/` from their phone without
    typing 'iGlyph.mobileconfig' on a touchscreen
    """

    profile_bytes: bytes = b""
    profile_name: str = "iGlyph.mobileconfig"
    on_download: "threading.Event | None" = None

    def _send_profile_headers(self) -> None:
        self.send_response(200)
        self.send_header("Content-Type", "application/x-apple-aspen-config")
        self.send_header("Content-Length", str(len(self.profile_bytes)))
        self.send_header(
            "Content-Disposition",
            f'attachment; filename="{self.profile_name}"',
        )
        self.end_headers()

    def do_HEAD(self) -> None:  # noqa: N802
        self._send_profile_headers()

    def do_GET(self) -> None:  # noqa: N802
        self._send_profile_headers()
        self.wfile.write(self.profile_bytes)
        if self.on_download is not None:
            self.on_download.set()

    def log_message(self, fmt: str, *args: object) -> None:
        ts = time.strftime("%H:%M:%S")
        sys.stderr.write(f"[{ts}] {self.address_string()} {fmt % args}\n")


class ReusableThreadingTCPServer(socketserver.ThreadingTCPServer):
    """ThreadingTCPServer with SO_REUSEADDR so re-runs after Ctrl-C don't
    hit 'address already in use' during the TIME_WAIT window."""

    allow_reuse_address = True


def serve_profile(
    profile_bytes: bytes,
    profile_name: str,
    port: int,
    timeout_minutes: int,
) -> None:
    """Spin up an HTTP server, print URLs, exit on download or timeout"""
    MobileConfigHandler.profile_bytes = profile_bytes
    MobileConfigHandler.profile_name = profile_name
    download_event = threading.Event()
    MobileConfigHandler.on_download = download_event

    # ThreadingTCPServer so the response actually completes before shutdown
    server = ReusableThreadingTCPServer(("0.0.0.0", port), MobileConfigHandler)
    server.daemon_threads = True
    actual_port = server.server_address[1]

    ips = get_local_ips()

    print(f"\nserving {profile_name} ({len(profile_bytes):,} bytes) on all interfaces")
    print("open one of these on your iPhone/iPad (Safari):")
    for ip in ips:
        print(f"  http://{ip}:{actual_port}/")
    print(
        f"\nwaiting for download "
        f"(timeout: {timeout_minutes}m, exits on first download, Ctrl-C to cancel)"
    )

    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
    server_thread.start()

    try:
        triggered = download_event.wait(timeout=timeout_minutes * 60)
        if triggered:
            time.sleep(0.5)  # let the response finish writing
            print("download served - exiting cleanly")
        else:
            print(f"\ntimeout reached ({timeout_minutes}m), no download - exiting")
            sys.exit(1)
    except KeyboardInterrupt:
        print("\ncancelled by user")
        sys.exit(130)
    finally:
        server.shutdown()
        server.server_close()


# =======================================================================
# CLI
# =======================================================================


def format_listing(annotated: list[tuple[Path, str | None]]) -> str:
    """Pretty-print the font listing for --list / --dry-run"""
    lines = [f"would bundle {len(annotated)} font(s):"]
    total_bytes = 0
    for path, ps in annotated:
        size = path.stat().st_size
        total_bytes += size
        ps_display = ps if ps else "<no PS name found>"
        lines.append(f"  {path.name:40s}  {size:>8,} B  PS: {ps_display}")
    lines.append(f"total payload: {total_bytes:,} bytes")
    return "\n".join(lines)


def _port_number(value: str) -> int:
    """argparse type: accept 0-65535, reject everything else."""
    try:
        port = int(value)
    except ValueError:
        raise argparse.ArgumentTypeError(f"port must be an integer, got {value!r}")
    if not 0 <= port <= 65535:
        raise argparse.ArgumentTypeError(f"port must be 0-65535, got {port}")
    return port


def _positive_int(value: str) -> int:
    """argparse type: accept positive integers only."""
    try:
        n = int(value)
    except ValueError:
        raise argparse.ArgumentTypeError(f"must be a positive integer, got {value!r}")
    if n <= 0:
        raise argparse.ArgumentTypeError(f"must be > 0, got {n}")
    return n


def main() -> int:
    p = argparse.ArgumentParser(
        prog="iglyph",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument(
        "inputs",
        nargs="+",
        type=Path,
        metavar="INPUT",
        help="font files or directories containing .ttf/.otf files",
    )
    p.add_argument(
        "-o",
        "--output",
        type=Path,
        default=Path("iGlyph.mobileconfig"),
        help="output .mobileconfig path (default: iGlyph.mobileconfig)",
    )
    p.add_argument(
        "-l",
        "--list",
        "-d",
        "--dry-run",
        dest="list_only",
        action="store_true",
        help="list what would be bundled without writing anything",
    )
    p.add_argument(
        "--name",
        default="iGlyph Fonts",
        help="profile display name (default: 'iGlyph Fonts')",
    )
    p.add_argument(
        "--identifier",
        default=None,
        help=(
            "reverse-DNS profile identifier (default: auto-derived from "
            "--name + content hash, so distinct font bundles coexist on device)"
        ),
    )
    p.add_argument(
        "--organization",
        default="iGlyph",
        help="organization name shown on install (default: 'iGlyph')",
    )
    p.add_argument(
        "--serve",
        nargs="?",
        const=18888,
        default=None,
        type=_port_number,
        metavar="PORT",
        help="serve the profile over HTTP (default port 18888)",
    )
    p.add_argument(
        "--timeout",
        type=_positive_int,
        default=10,
        metavar="MINUTES",
        help="serve timeout in minutes (default: 10)",
    )

    args = p.parse_args()

    # Discover and validate
    try:
        fonts = discover_fonts(args.inputs)
        valid_fonts, annotated = validate_and_report(fonts)
    except FontError as e:
        print(f"error: {e}", file=sys.stderr)
        return 1

    # List mode exits here
    if args.list_only:
        print(format_listing(annotated))
        return 0

    # Build profile
    identifier = args.identifier or derive_identifier(args.name, annotated)
    blob = build_profile(
        annotated,
        name=args.name,
        identifier=identifier,
        organization=args.organization,
    )

    # Write output, with cleaner errors than a raw OSError traceback
    try:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_bytes(blob)
    except OSError as e:
        print(f"error: could not write {args.output}: {e}", file=sys.stderr)
        return 1
    print(f"wrote {args.output} ({len(blob):,} bytes, {len(valid_fonts)} font(s))")

    # Serve mode
    if args.serve is not None:
        serve_profile(
            profile_bytes=blob,
            profile_name=args.output.name,
            port=args.serve,
            timeout_minutes=args.timeout,
        )
        return 0

    # Otherwise, print install hints
    print("\nfastest path:")
    print(f"  python3 iglyph.py {' '.join(str(i) for i in args.inputs)} --serve")
    print("\nmanual path:")
    print(f"  1. AirDrop {args.output} to the device")
    print("  2. Open it -> a Settings prompt appears")
    print("  3. Settings -> General -> VPN & Device Management -> install")
    print("  4. fonts appear in Settings -> General -> Fonts and in font pickers")
    return 0


if __name__ == "__main__":
    sys.exit(main())