path: root/packages/meshbay-node/src/meshbay_node/config.py

"""
MeshBay Node configuration.

Config file: ~/.config/meshbay/node.toml
All values have sensible defaults and can be overridden by env vars
prefixed with MESHBAY_ (e.g. MESHBAY_HUB_URL).
"""

import logging
import os
from dataclasses import dataclass, field
from pathlib import Path

from meshbay_node.platform import config_dir, data_dir

try:
    import tomllib  # Python 3.11+
except ImportError:
    import tomli as tomllib  # type: ignore[no-redef]

log = logging.getLogger(__name__)

DEFAULT_STUN_SERVERS: list[str] = [
    "stun:stun.l.google.com:19302",
    "stun:stun1.l.google.com:19302",
    "stun:stun.cloudflare.com:3478",
    # stun.services.mozilla.com was here — Mozilla retired it, the name no
    # longer resolves, and each gather waited out its DNS timeout. Two
    # providers (Google, Cloudflare) still cover a single-provider outage.
]

DEFAULT_CONFIG_PATH = config_dir() / "node.toml"

EXAMPLE_CONFIG = """\
# MeshBay Node configuration — multi-group example
# See: https://meshbay.org/docs/node-config

[hub]
url      = "https://meshbay.org"
username = "myusername"

[node]
# QUIC (MNP) direct path — LAN, port-forwarded, hub-less. Off by default: no
# client speaks QUIC yet, so leaving it on only opens a UDP port.
quic_enabled = false
quic_port = 19010
ui_port  = 18000   # local control API — JSON, 127.0.0.1 only, token-gated

# One-time codes. An invitation waits for someone to read their messages; an
# operator pairing code is typed during the SSH session that printed it.
invite_ttl_hours = 168   # 7 days
pair_ttl_hours   = 24
# How long a new device may wait for one of your existing devices to approve it.
device_request_ttl_minutes = 60

# How many people may watch a video at once. One ffmpeg runs per viewer for as
# long as they watch — it remuxes rather than re-encodes, so it costs little CPU
# and around 50 MB of memory, and spends most of the film idle. Past this, a
# viewer is told the server is busy. Raise it on a machine with memory to spare;
# lower it on a Pi.
max_concurrent_streams = 8

# How many downloads and uploads run at once on this node, across every group.
# A slot is concurrency, not bandwidth: what it protects is open file handles,
# disk seeks and the channel buffer each transfer keeps full. Past this, a
# member is queued and told so, and starts when a slot frees.
max_concurrent_downloads = 8
max_concurrent_uploads = 8

# The largest single file a member may upload to this node, in GB. It is this
# machine's disk that fills, so the ceiling is the operator's to set: lower it
# on a small disk, raise it for a library of films. Fractions are allowed
# (0.5 = 512 MB). A file past it is refused at the chunk that crosses it and
# the partial file is deleted.
max_upload_gb = 8

# HEVC sources have no browser decoder on most platforms, so streaming one is
# transcoded to H264 rather than the usual free copy — real CPU per viewer.
# Set to false only if every viewer's client is known to decode HEVC itself.
transcode_incompatible_video = true

# Use the GPU for that re-encode when there is one that works — VA-API here,
# Quick Sync or NVENC on Windows. Established by encoding 1080p and checking
# the result, never guessed from the hardware, and ignored where the test
# fails: this is on by default and the only reason to turn it off is a driver
# that misbehaves in a way the test does not catch. On an Intel mini-PC it is
# the difference between transcoding in real time and having to set
# transcode_incompatible_video = false.
hardware_video_encode = true

# ICE candidate gathering. By default, virtual/VPN interfaces (Tailscale,
# libvirt, Docker) are auto-excluded — a STUN request that can't reach the
# server holds the WebRTC answer for 5 seconds.  Set this to restrict
# gathering to specific interfaces.  An entry matches the OS adapter name, the
# device description, or one of the adapter's own IPv4 addresses, so a Windows
# host can be named by description or address rather than by its GUID.  An
# entry that matches nothing is ignored with a warning rather than leaving the
# node with no candidates at all.
# ice_interfaces = ["wlp0s20f3", "eth0", "192.168.1.22"]

# STUN servers for WebRTC ICE candidate gathering (NAT traversal). By default
# four public servers are used; set this to override. Every server in the list
# is queried in parallel each gather and the first answer wins, so one slow or
# blocked server no longer stalls the WebRTC answer.
# stun_servers = ["stun:stun.l.google.com:19302", "stun:stun.cloudflare.com:3478"]

# Browser and native clients reach this node over WebRTC DataChannel via hub
# signaling — no inbound port to open. QUIC is the optional direct path.

# Multiple groups — each with one or more named directories ("roots").
#
# A root's name is the directory's basename, and it becomes the first segment of
# every path members see: /home/user/Media appears to everyone as "Media/".
# Two roots cannot share a name (compared without regard to case), and no root
# may sit inside another. A writable root accepts uploads from group members.
[[groups]]
id          = ""    # set after joining
name        = "My Media"
quic_port   = 19010

  [[groups.roots]]
  path     = "/home/user/Media"
  writable = true

  [[groups.roots]]
  path      = "/run/media/user/USB/Musique"
  kind      = "audio"
  removable = true   # eject before unplugging

# The single-directory form still works and means the same thing — one root,
# named after the directory, receiving uploads.
[[groups]]
id          = ""
name        = "Public Archive"
shared_dir  = "/home/user/Archive"
quic_port   = 19012
visibility  = "public"       # discoverable on the hub
# join_policy = "open"       # anyone the hub says is a member gets the group key,
                             # with no pairing code. Only for groups where that is
                             # genuinely intended: it means the hub can join too.

[keystore]
# unlock_file = "~/.config/meshbay/unlock.key"
# or set MESHBAY_UNLOCK_KEY env var

# Operator authority is not configured here. Run `meshbay-node operator pair` and
# enter the code in your browser: the node pins that browser's key, and invites
# and file deletion are signed with it.
"""


@dataclass
class HubConfig:
    url:      str = "https://meshbay.org"
    username: str = ""


@dataclass
class NodeConfig:
    quic_port:  int = 19010
    # The QUIC MNP listener. Off by default: no shipping client speaks QUIC yet
    # (the browser and the desktop client use WebRTC; the hub-less `group://`
    # sidecar is unbuilt), so starting it only opens a UDP port with nothing to
    # reach it. Turn on for LAN / port-forwarded / hub-less direct access once a
    # client for it exists. `punch_nat()` is a direct-connection helper, not a
    # NAT-traversal stack — a peer behind NAT still needs the port forwarded.
    quic_enabled: bool = False
    ui_port:    int = 18000
    # How long a one-time code stays usable. Invitations travel through a human
    # conversation and are answered days later; operator pairing happens during
    # the SSH session that printed it.
    invite_ttl_hours: int = 168   # 7 days
    pair_ttl_hours:   int = 24
    # A device-add code is read off one screen and typed into another, in one
    # sitting. Comfort rather than security: the code is bound to the requesting
    # keys by its hash, so a longer window widens nothing an attacker can use.
    device_request_ttl_minutes: int = 60
    # How many people may watch a video at the same time. One ffmpeg runs per
    # viewer for as long as they watch, so this is the knob that decides when
    # the node answers "server busy" — see MAX_CONCURRENT_TRANSCODES in
    # transport/webrtc/apps/streaming.py for what one costs.
    max_concurrent_streams: int = 8
    # How many transfers run at once on this node, across every group —
    # separate pools, because a download and an upload cost different things
    # and one queue for both makes each cap meaningless. Streaming has its own
    # third pool (max_concurrent_streams above): a member watching a film is
    # not charged a download slot, and a download does not make the next film
    # answer "server busy". See meshbay_node/transfers.py.
    max_concurrent_downloads: int = 8
    max_concurrent_uploads: int = 8
    # The per-file upload ceiling, in GB — the one limit here that bounds a
    # member's writes to the operator's disk rather than this node's own
    # concurrency. There is still no aggregate quota (MESHBAY_DESIGN.md
    # §15.3), so this is what stands between a writable root and a full disk.
    max_upload_gb: float = 8.0
    # HEVC (and any future codec in media_probe.py's
    # BROWSER_INCOMPATIBLE_VIDEO_CODECS) has no decoder in most browsers, so
    # streaming it needs a real re-encode to H264 rather than the usual free
    # copy. On by default since the alternative is a hard "codec not
    # supported" error; set to false if this node's viewers are all clients
    # that already decode the source codec directly, since transcoding costs
    # real CPU per concurrent viewer, unlike the copy path.
    transcode_incompatible_video: bool = True
    # Whether that re-encode may run on the GPU. See hwaccel.py: the capability
    # is established by encoding and reading the result back, so `true` here
    # means "use it if it works", never "assume it does". Off returns the node
    # to libx264 on every stream, which is where it was before hardware
    # encoding existed.
    hardware_video_encode: bool = True
    # ICE candidate gathering: which network interfaces to include or exclude.
    # By default, virtual and VPN interfaces (Tailscale, libvirt, Docker) are
    # auto-excluded because a STUN request that can't reach the server holds
    # the gather for the full 5-second timeout — measured at 6 s total on a
    # machine with a Tailscale wt0 interface.
    ice_interfaces: list[str] = field(default_factory=list)  # include-list overrides auto
    stun_servers: list[str] = field(default_factory=list)  # empty = DEFAULT_STUN_SERVERS
    ffmpeg_path:  str = "ffmpeg"
    ffprobe_path: str = "ffprobe"


@dataclass
class RootSpec:
    """One named directory inside a group. See `meshbay_node/roots.py`."""
    path:   str = ""
    name:   str = ""          # empty → the directory's basename, derived at load
    kind:   str = "generic"   # generic|video|audio|photo — a view hint, unused for now
    writable:  bool = False   # RW roots accept uploads from group members
    removable: bool = False   # operator can eject this root before unplugging the device


@dataclass
class GroupConfig:
    id:         str = ""
    name:       str = ""
    # A group's content is several named roots. `shared_dir` is the single-root
    # form and is still read: it becomes one root named after its basename, which
    # is why every path gained a segment. See roots.py for why there is no
    # unprefixed shape.
    roots:      list[RootSpec] = field(default_factory=list)
    shared_dir: str = ""      # legacy single-root form, migrated at load
    upload_dir: str = ""      # legacy — migrated to a writable root
    visibility: str = "private"   # public|private — discoverability, not admission
    # Admission. "invite" (default) means a newcomer needs a one-time pairing code
    # before the node wraps the group key for them; "open" means the node pins
    # whoever turns up first (TOFU) and serves them.
    #
    # Deliberately read from THIS file and never from the hub: a hub that could
    # declare a group open would walk into any group it liked. Being findable
    # (`visibility`) and being open (`join_policy`) are different questions.
    join_policy: str = "invite"   # invite|open
    quic_port:  int = 19010       # QUIC MNP port

    def __post_init__(self) -> None:
        """
        The single-directory form becomes one root, whoever built this.

        On the dataclass rather than in the TOML reader, because a GroupConfig is
        also built by the CLI, by `group attach` and by tests. Putting the
        migration in the parser alone left every one of those paths with a group
        that had no directory at all — and it presented as "skipping group",
        which reads like configuration rather than a bug.
        """
        if not self.roots and self.shared_dir.strip():
            self.roots = [RootSpec(path=self.shared_dir.strip(), writable=True)]
        if self.upload_dir.strip():
            for r in self.roots:
                r.writable = False
            self.roots.append(RootSpec(
                path=self.upload_dir.strip(), writable=True))


@dataclass
class KeystoreConfig:
    path:        Path = field(default_factory=lambda: DEFAULT_CONFIG_PATH.parent / "keystore.enc")
    unlock_file: Path | None = None


@dataclass
class Config:
    hub:      HubConfig      = field(default_factory=HubConfig)
    node:     NodeConfig     = field(default_factory=NodeConfig)
    groups:   list[GroupConfig] = field(default_factory=list)
    keystore: KeystoreConfig = field(default_factory=KeystoreConfig)
    data_dir: Path = field(default_factory=data_dir)

    # Back-compat: single-group access
    @property
    def group(self) -> GroupConfig:
        return self.groups[0] if self.groups else GroupConfig()


def _positive(value: object, default: int, name: str) -> int:
    """A count that must be at least one, or the default with a word about it.

    Zero is the dangerous one: `asyncio.Semaphore(0)` is not "no limit", it is
    a node where no video ever plays and nothing in the log says why.
    """
    # bool before int: TOML `true` is a bool, and `int(True)` is 1 — a node
    # where exactly one person may watch, arrived at by a typo and announced
    # nowhere.
    if isinstance(value, bool) or not isinstance(value, (int, str)):
        log.warning("%s = %r is not a count — using %d", name, value, default)
        return default
    try:
        n = int(value)
    except (TypeError, ValueError):
        log.warning("%s = %r is not a number — using %d", name, value, default)
        return default
    if n < 1:
        log.warning("%s = %d would stop the feature entirely — using %d",
                    name, n, default)
        return default
    return n


def _positive_float(value: object, default: float, name: str) -> float:
    """A size that must be greater than zero, or the default with a word about it.

    Separate from `_positive` because this one is a quantity, not a count: half
    a gigabyte is a legitimate ceiling on a small disk, and rounding it to zero
    would refuse every upload with nothing in the log to say why.
    """
    if isinstance(value, bool) or not isinstance(value, (int, float, str)):
        log.warning("%s = %r is not a size — using %g", name, value, default)
        return default
    try:
        n = float(value)
    except (TypeError, ValueError):
        log.warning("%s = %r is not a number — using %g", name, value, default)
        return default
    if n <= 0:
        log.warning("%s = %g would refuse every upload — using %g",
                    name, n, default)
        return default
    return n


def _read_roots(group: dict) -> list[RootSpec]:
    """
    A group's roots, from `[[groups.roots]]` or from the legacy `shared_dir`.

    Both forms are accepted and `shared_dir` is not deprecated for a single
    directory — it is the same thing said shorter. Naming both is refused rather
    than merged: which one receives uploads would be a guess, and a wrong guess
    is discovered weeks later.
    """
    specs = []
    for r in group.get("roots", []) or []:
        # Backward compat: old configs have `upload = true` instead of `writable`
        writable = bool(r.get("writable", r.get("upload", False)))
        specs.append(RootSpec(
            path=str(r.get("path", "")),
            name=str(r.get("name", "")),
            kind=str(r.get("kind", "generic")),
            writable=writable,
            removable=bool(r.get("removable", False)),
        ))
    legacy = str(group.get("shared_dir", "") or "").strip()
    if specs and legacy:
        log.warning(
            "group %r declares both shared_dir and [[groups.roots]] — using "
            "roots and ignoring shared_dir = %s",
            group.get("name", ""), legacy)
    # A bare shared_dir needs no handling here: GroupConfig.__post_init__ turns
    # it into one root for every construction path, not just this one.
    return specs


def node_settings_defaults(nd: NodeConfig | None = None) -> dict:
    """
    The `node.toml` side of every setting the roster resolves.

    One function because there were three copies of this dict written by hand
    and they disagreed. The daemon's left out both transfer pools, so on a node
    whose operator had never touched the panel they resolved to None, were
    written back onto the config, and the transport fell through to its own
    defaults — `node.toml` parsed, validated, and then ignored.

    `test_node_settings_defaults.py` holds this against
    `Roster.node_setting_keys()`, because a key missing here raises nothing
    anywhere: it is a setting that stops working quietly.
    """
    nd = nd or NodeConfig()
    return {
        "invite_ttl_hours": nd.invite_ttl_hours,
        "pair_ttl_hours": nd.pair_ttl_hours,
        "device_request_ttl_minutes": nd.device_request_ttl_minutes,
        "max_concurrent_streams": nd.max_concurrent_streams,
        "max_concurrent_downloads": nd.max_concurrent_downloads,
        "max_concurrent_uploads": nd.max_concurrent_uploads,
        "max_upload_gb": nd.max_upload_gb,
        "transcode_incompatible_video": nd.transcode_incompatible_video,
        "stun_servers": nd.stun_servers or list(DEFAULT_STUN_SERVERS),
        "ice_interfaces": nd.ice_interfaces,
    }


def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
    """
    Load config from TOML file. Supports both single [group] and
    multiple [[groups]] sections. Env vars override file values.
    """
    cfg = Config()

    if path.exists():
        raw = tomllib.loads(path.read_text(encoding="utf-8"))

        hub = raw.get("hub", {})
        cfg.hub.url      = hub.get("url",      cfg.hub.url)
        cfg.hub.username = hub.get("username", cfg.hub.username)

        nd = raw.get("node", {})
        # `port` (TCP+TLS) and `http_port` no longer exist — both listeners were removed
        # in Phase 11.5 (findings C1, C6). Regenerate node.toml with `meshbay-node init`.
        cfg.node.quic_port = nd.get("quic_port", cfg.node.quic_port)
        cfg.node.quic_enabled = bool(nd.get("quic_enabled", cfg.node.quic_enabled))
        cfg.node.ui_port   = nd.get("ui_port",   cfg.node.ui_port)
        cfg.node.invite_ttl_hours = int(
            nd.get("invite_ttl_hours", cfg.node.invite_ttl_hours))
        cfg.node.pair_ttl_hours = int(
            nd.get("pair_ttl_hours", cfg.node.pair_ttl_hours))
        cfg.node.device_request_ttl_minutes = int(
            nd.get("device_request_ttl_minutes",
                   cfg.node.device_request_ttl_minutes))
        cfg.node.max_concurrent_streams = _positive(
            nd.get("max_concurrent_streams", cfg.node.max_concurrent_streams),
            cfg.node.max_concurrent_streams, "max_concurrent_streams")
        cfg.node.max_concurrent_downloads = _positive(
            nd.get("max_concurrent_downloads", cfg.node.max_concurrent_downloads),
            cfg.node.max_concurrent_downloads, "max_concurrent_downloads")
        cfg.node.max_concurrent_uploads = _positive(
            nd.get("max_concurrent_uploads", cfg.node.max_concurrent_uploads),
            cfg.node.max_concurrent_uploads, "max_concurrent_uploads")
        cfg.node.max_upload_gb = _positive_float(
            nd.get("max_upload_gb", cfg.node.max_upload_gb),
            cfg.node.max_upload_gb, "max_upload_gb")
        cfg.node.transcode_incompatible_video = bool(
            nd.get("transcode_incompatible_video", cfg.node.transcode_incompatible_video))
        cfg.node.hardware_video_encode = bool(
            nd.get("hardware_video_encode", cfg.node.hardware_video_encode))
        ice_if = nd.get("ice_interfaces")
        if isinstance(ice_if, list):
            cfg.node.ice_interfaces = [str(s) for s in ice_if]
        stun = nd.get("stun_servers")
        if isinstance(stun, list):
            cfg.node.stun_servers = [str(s) for s in stun]
        if "ffmpeg_path" in nd:
            cfg.node.ffmpeg_path = str(nd["ffmpeg_path"])
        if "ffprobe_path" in nd:
            cfg.node.ffprobe_path = str(nd["ffprobe_path"])

        # Multi-group: [[groups]] array
        if "groups" in raw:
            for g in raw["groups"]:
                cfg.groups.append(GroupConfig(
                    id=g.get("id", ""),
                    name=g.get("name", ""),
                    roots=_read_roots(g),
                    # Ignored when roots are given explicitly (warned about in
                    # _read_roots); otherwise __post_init__ migrates it.
                    shared_dir="" if _read_roots(g) else g.get("shared_dir", ""),
                    upload_dir=g.get("upload_dir", ""),
                    visibility=g.get("visibility", "private"),
                    join_policy=g.get("join_policy", "invite"),
                    quic_port=g.get("quic_port", cfg.node.quic_port),
                ))
        # Back-compat: single [group] section
        elif "group" in raw:
            grp = raw["group"]
            cfg.groups.append(GroupConfig(
                id=grp.get("id", ""),
                name=grp.get("name", ""),
            ))

        if "data_dir" in raw:
            cfg.data_dir = Path(raw["data_dir"]).expanduser().resolve()

        if "admin_pk_ed25519" in raw:
            # Removed, not merely unused: a key named here granted operator
            # authority, and dropping it silently would refuse invites and file
            # deletion with a signature error that looks like something else.
            log.warning(
                "admin_pk_ed25519 in %s is ignored — operator authority now comes "
                "from the roster. Run `meshbay-node operator pair` and delete the "
                "line.", path)

        ks = raw.get("keystore", {})
        if "path" in ks:
            cfg.keystore.path = Path(ks["path"]).expanduser()
        if "unlock_file" in ks:
            cfg.keystore.unlock_file = Path(ks["unlock_file"]).expanduser()

    # Env var overrides
    if url  := os.environ.get("MESHBAY_HUB_URL"):
        cfg.hub.url = url
    if user := os.environ.get("MESHBAY_USERNAME"):
        cfg.hub.username = user
    if port := os.environ.get("MESHBAY_QUIC_PORT"):
        cfg.node.quic_port = int(port)
    if (qe := os.environ.get("MESHBAY_QUIC_ENABLED")) is not None:
        cfg.node.quic_enabled = qe.strip().lower() in ("1", "true", "yes", "on")
    if streams := os.environ.get("MESHBAY_MAX_CONCURRENT_STREAMS"):
        cfg.node.max_concurrent_streams = _positive(
            streams, cfg.node.max_concurrent_streams,
            "MESHBAY_MAX_CONCURRENT_STREAMS")

    return cfg


def write_example_config(path: Path = DEFAULT_CONFIG_PATH) -> None:
    """Write an example config file if none exists."""
    if not path.exists():
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(EXAMPLE_CONFIG, encoding="utf-8", newline="\n")