mc-webui/app/demo_guard.py

"""
Demo mode — read-only guard for a publicly shared instance.

mc-webui has no authentication: whoever reaches the port can rename the radio,
rewrite its frequency, wipe the address book or type into the console. That is
fine on a home LAN and not fine once the instance is handed to strangers, which
is what MC_DEMO=true is for.

The rules, in the order they are applied:

1. MC_DEMO off  → nothing changes, every request passes. Existing installs are
   untouched by this module.
2. Caller unlocked → every request passes. A caller is unlocked by coming from
   a network in MC_DEMO_TRUSTED_NETS, or by holding the cookie handed out for
   MC_DEMO_UNLOCK_CODE.
3. Otherwise → writes are refused (default deny, see WRITE_ALLOWLIST for the
   handful of things a visitor may still do), a few reads that hand out secrets
   are refused too (READ_DENYLIST), and the console accepts only the commands
   in CONSOLE_READONLY.

Default deny is the point. New endpoints are locked the day they are written
rather than the day someone remembers to add them to a list.
"""

import hmac
import ipaddress
import logging
import re
from hashlib import sha256
from typing import List, Optional

from app.config import config

logger = logging.getLogger(__name__)

# Cookie carrying the unlock proof. HttpOnly — the UI learns whether it is
# unlocked from /api/status, never by reading this back.
UNLOCK_COOKIE = 'mc_demo_unlock'
UNLOCK_COOKIE_MAX_AGE = 30 * 24 * 3600  # 30 days

# Domain separation, so the token cannot be replayed as some other HMAC.
_UNLOCK_MESSAGE = b'mc-webui-demo-unlock-v1'

# Error payload for a refused call. The frontend greys these controls out, so
# seeing this means either a stale page or a direct API call.
DEMO_ERROR = 'demo_locked'


# ================================================================
# What stays reachable while locked
# ================================================================

# Writes a visitor may still perform. Anything not listed here is refused.
# Chatting is the whole point of the demo, so sending stays open; every form
# of configuration does not.
WRITE_ALLOWLIST = [
    re.compile(r'^/api/messages$'),
    re.compile(r'^/api/messages/\d+/resend$'),
    re.compile(r'^/api/dm/messages$'),
    re.compile(r'^/api/read_status/mark_read$'),
    re.compile(r'^/api/read_status/mark_all_read$'),
    # The unlock endpoints themselves, or there is no way back in.
    re.compile(r'^/api/demo/unlock$'),
    re.compile(r'^/api/demo/lock$'),
]

# Reads refused while locked. These are GETs, so the write guard never sees
# them, and each one hands out something a visitor has no business holding:
# stored repeater admin passwords, a copy of the whole database, captured
# message text, or the server log.
READ_DENYLIST = [
    re.compile(r'^/api/repeaters/[^/]+/password$'),
    re.compile(r'^/api/backup/list$'),
    re.compile(r'^/api/backup/download$'),
    re.compile(r'^/api/diagnostics/captures/[^/]+/download$'),
    re.compile(r'^/api/logs$'),
]

# Console commands that only read. Everything else — every `set`, every form of
# transmit, every delete — is refused. `get_channel` is absent on purpose: it
# prints a channel's shared key. `clock` is absent because the same branch also
# sets the device clock.
CONSOLE_READONLY = frozenset({
    'infos', 'status', 'stats', 'bat', 'ver', 'help',
    'contacts', 'contacts_all', 'contact_info', 'pending_contacts',
    'channels', 'path', 'get',
})


# ================================================================
# Trusted networks
# ================================================================

def _parse_nets(raw: str) -> List[ipaddress._BaseNetwork]:
    """Parse the comma-separated CIDR list, skipping (and logging) bad entries."""
    nets = []
    for chunk in raw.split(','):
        chunk = chunk.strip()
        if not chunk:
            continue
        try:
            nets.append(ipaddress.ip_network(chunk, strict=False))
        except ValueError:
            logger.warning("MC_DEMO_TRUSTED_NETS: ignoring unparseable entry %r", chunk)
    return nets


# Parsed once per distinct value of the setting rather than once per import, so
# the parse stays off the request path without freezing the config at import
# time — which is what lets the tests drive it.
_nets_cache: tuple = (None, [])   # (raw value it was parsed from, parsed nets)


def trusted_nets() -> List[ipaddress._BaseNetwork]:
    """The configured trusted networks, parsed."""
    global _nets_cache
    raw = config.MC_DEMO_TRUSTED_NETS or ''
    if _nets_cache[0] != raw:
        _nets_cache = (raw, _parse_nets(raw))
    return _nets_cache[1]


def _came_through_a_proxy(req) -> bool:
    """
    True when the request carries a hop header, i.e. the peer address belongs to
    a proxy or tunnel rather than to the client.
    """
    return any(
        h in req.headers
        for h in ('X-Forwarded-For', 'X-Real-IP', 'Forwarded', 'CF-Connecting-IP')
    )


def ip_is_trusted(req) -> bool:
    """
    True when the caller's address falls inside MC_DEMO_TRUSTED_NETS.

    The subtlety worth keeping: when the instance is published through a tunnel
    (cloudflared, ngrok, a VPN box) every visitor arrives from that tunnel's
    address. If the tunnel runs on the LAN, that address sits inside the very
    network the operator listed as trusted — and matching on it would unlock the
    instance for the whole internet at once. So an address is only accepted when
    it is genuinely the client's: either no hop header is present, or
    MC_TRUST_PROXY is on and ProxyFix has already replaced remote_addr with the
    real client address.
    """
    nets = trusted_nets()
    if not nets:
        return False
    if not config.MC_TRUST_PROXY and _came_through_a_proxy(req):
        return False

    addr = req.remote_addr
    if not addr:
        return False
    try:
        ip = ipaddress.ip_address(addr.split('%')[0])  # strip any zone id
    except ValueError:
        return False
    return any(ip in net for net in nets)


# ================================================================
# Unlock code
# ================================================================

def unlock_token() -> str:
    """The cookie value proving knowledge of MC_DEMO_UNLOCK_CODE."""
    return hmac.new(
        config.MC_DEMO_UNLOCK_CODE.encode('utf-8'), _UNLOCK_MESSAGE, sha256
    ).hexdigest()


def code_is_valid(code: Optional[str]) -> bool:
    """Constant-time check of a code typed by the user."""
    if not config.MC_DEMO_UNLOCK_CODE or not code:
        return False
    return hmac.compare_digest(code, config.MC_DEMO_UNLOCK_CODE)


def cookie_is_valid(req) -> bool:
    """Constant-time check of the unlock cookie."""
    if not config.MC_DEMO_UNLOCK_CODE:
        return False
    presented = req.cookies.get(UNLOCK_COOKIE)
    if not presented:
        return False
    return hmac.compare_digest(presented, unlock_token())


# ================================================================
# The decision
# ================================================================

def is_demo() -> bool:
    """True when this instance runs in demo mode at all."""
    return bool(config.MC_DEMO)


def is_unlocked(req) -> bool:
    """
    True when the caller may do anything — because demo mode is off, or because
    they proved they are the operator.
    """
    if not config.MC_DEMO:
        return True
    return ip_is_trusted(req) or cookie_is_valid(req)


def _matches(path: str, patterns) -> bool:
    return any(p.match(path) for p in patterns)


def request_is_blocked(req) -> bool:
    """
    True when this request must be refused. Call it from a before_request hook.
    """
    if is_unlocked(req):
        return False

    path = req.path.rstrip('/') or '/'

    if req.method in ('GET', 'HEAD', 'OPTIONS'):
        return _matches(path, READ_DENYLIST)

    return not _matches(path, WRITE_ALLOWLIST)


def console_command_is_blocked(command: str) -> bool:
    """
    True when a console command must be refused. `command` is the raw line; only
    its first word decides, which is the same thing the router dispatches on.
    """
    if not config.MC_DEMO:
        return False
    head = (command or '').strip().split()
    if not head:
        return False
    return head[0].lower() not in CONSOLE_READONLY


# ================================================================
# Startup reporting
# ================================================================

def log_startup_state() -> None:
    """
    Say plainly what the guard will do, once, at boot. A demo instance that
    quietly failed to lock looks exactly like one that locked correctly, so the
    log is the only place the difference shows up before a tester finds it.
    """
    if not config.MC_DEMO:
        return

    logger.info("Demo mode ON — writes are refused unless the caller is unlocked")

    nets = trusted_nets()
    if nets:
        logger.info(
            "Demo mode: trusted networks %s",
            ', '.join(str(n) for n in nets),
        )
        if not config.MC_TRUST_PROXY:
            logger.warning(
                "Demo mode: MC_DEMO_TRUSTED_NETS is set but MC_TRUST_PROXY is off. "
                "Requests arriving with a proxy/tunnel header will NOT be trusted by "
                "address, because that address would be the proxy's and not the "
                "visitor's. Set MC_TRUST_PROXY=true only if the app is reached "
                "exclusively through your proxy."
            )
    else:
        logger.info("Demo mode: no trusted networks configured")

    if config.MC_DEMO_UNLOCK_CODE:
        if len(config.MC_DEMO_UNLOCK_CODE) < 8:
            logger.warning(
                "Demo mode: MC_DEMO_UNLOCK_CODE is shorter than 8 characters — "
                "it is the only thing standing between a visitor and the radio"
            )
        else:
            logger.info("Demo mode: unlock code is set")
    else:
        logger.info("Demo mode: no unlock code set")

    if not nets and not config.MC_DEMO_UNLOCK_CODE:
        logger.warning(
            "Demo mode: neither MC_DEMO_TRUSTED_NETS nor MC_DEMO_UNLOCK_CODE is "
            "set — this instance is locked for everyone, including you"
        )