Discover DX media consent / DMC10F1 devices on the local network.

discover_dmc10f1.py

#!/usr/bin/env python3
"""Discover DX media consent / DMC10F1 devices on the local network.

The official app sends an SSDP-like probe to 239.255.255.250:61900:

    M-SEARCH * HTTP:/1.1
    HOST: 239.255.255.250:61900
    MAN: "ssdp:discover"
    MX: 5
    ST: dx-media-consent

This script reproduces that probe and prints any UDP replies.
"""

from __future__ import annotations

import argparse
import ipaddress
import socket
import sys
import time
from dataclasses import dataclass


MULTICAST_ADDR = "239.255.255.250"
DISCOVERY_PORT = 61900
SEARCH_TARGET = "dx-media-consent"


@dataclass(frozen=True)
class Reply:
    address: str
    port: int
    payload: bytes


def build_probe(host: str, port: int, st: str, mx: int) -> bytes:
    # Match the firmware/app quirk: "HTTP:/1.1" has one slash, not two.
    lines = [
        "M-SEARCH * HTTP:/1.1",
        f"HOST: {host}:{port}",
        'MAN: "ssdp:discover"',
        f"MX: {mx}",
        f"ST: {st}",
        "",
        "",
    ]
    return "\r\n".join(lines).encode("ascii")


def parse_headers(payload: bytes) -> dict[str, str]:
    text = payload.rstrip(b"\x00").decode("utf-8", errors="replace")
    headers: dict[str, str] = {}
    for line in text.replace("\r\n", "\n").split("\n"):
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        headers[key.strip().lower()] = value.strip()
    return headers


def printable(payload: bytes) -> str:
    trimmed = payload.rstrip(b"\x00")
    return "".join(chr(b) if 32 <= b <= 126 or b in (9, 10, 13) else "." for b in trimmed)


def make_socket(args: argparse.Namespace) -> socket.socket:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    sock.settimeout(args.timeout)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    if hasattr(socket, "SO_REUSEPORT"):
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        except OSError:
            pass
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, args.ttl)

    if args.interface:
        try:
            ipaddress.ip_address(args.interface)
        except ValueError:
            raise SystemExit("--interface must be an IPv4 address, for example 192.168.1.10")
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(args.interface))

    bind_host = args.bind or ""
    try:
        sock.bind((bind_host, args.source_port))
    except OSError as exc:
        target = f"{bind_host or '0.0.0.0'}:{args.source_port}"
        raise SystemExit(
            f"Could not bind UDP {target}: {exc}. "
            "Close any app using the discovery port, or try --source-port 0 "
            "if you only need to send the probe."
        ) from exc

    return sock


def discover(args: argparse.Namespace) -> list[Reply]:
    probe = build_probe(args.group, args.port, args.st, args.mx)
    replies: dict[tuple[str, int, bytes], Reply] = {}

    with make_socket(args) as sock:
        if args.verbose:
            sys.stderr.write(probe.decode("ascii"))

        deadline = time.monotonic() + args.listen
        for attempt in range(args.count):
            sock.sendto(probe, (args.group, args.port))
            if args.verbose:
                sys.stderr.write(f"sent probe {attempt + 1}/{args.count}\n")

            per_attempt_deadline = min(deadline, time.monotonic() + args.interval)
            while time.monotonic() < per_attempt_deadline:
                try:
                    payload, (address, port) = sock.recvfrom(65535)
                except socket.timeout:
                    break
                replies[(address, port, payload)] = Reply(address, port, payload)

        while time.monotonic() < deadline:
            try:
                payload, (address, port) = sock.recvfrom(65535)
            except socket.timeout:
                break
            replies[(address, port, payload)] = Reply(address, port, payload)

    return sorted(replies.values(), key=lambda r: (r.address, r.port, r.payload))


def main() -> int:
    parser = argparse.ArgumentParser(description="Discover DMC10F1 / dx-media-consent devices.")
    parser.add_argument("--group", default=MULTICAST_ADDR, help=f"multicast address (default: {MULTICAST_ADDR})")
    parser.add_argument("--port", type=int, default=DISCOVERY_PORT, help=f"discovery port (default: {DISCOVERY_PORT})")
    parser.add_argument("--st", default=SEARCH_TARGET, help=f"search target (default: {SEARCH_TARGET})")
    parser.add_argument("--mx", type=int, default=5, help="MX value in the probe (default: 5)")
    parser.add_argument("--count", type=int, default=1, help="number of probes to send (default: 1)")
    parser.add_argument("--interval", type=float, default=0.5, help="seconds to wait after each probe (default: 0.5)")
    parser.add_argument("--listen", type=float, default=5.0, help="total seconds to listen for replies (default: 5)")
    parser.add_argument("--timeout", type=float, default=0.5, help="socket timeout in seconds (default: 0.5)")
    parser.add_argument("--ttl", type=int, default=1, help="multicast TTL (default: 1)")
    parser.add_argument("--interface", help="source interface IPv4 address for multicast")
    parser.add_argument("--bind", help="local IPv4 address to bind")
    parser.add_argument(
        "--source-port",
        type=int,
        default=DISCOVERY_PORT,
        help=f"local UDP source port/listen port (default: {DISCOVERY_PORT})",
    )
    parser.add_argument("--verbose", action="store_true", help="print the probe and send progress to stderr")
    args = parser.parse_args()

    replies = discover(args)
    if not replies:
        print("No dx-media-consent replies found.")
        return 1

    for index, reply in enumerate(replies, 1):
        print(f"[{index}] {reply.address}:{reply.port} ({len(reply.payload)} bytes)")
        headers = parse_headers(reply.payload)
        if headers:
            device_ip = headers.get("ip-address", reply.address)
            cmd_port = headers.get("cmd-port")
            ts_port = headers.get("ts-port")
            dev_ver = headers.get("dev-ver")
            app_ver = headers.get("app-ver")
            summary = [f"device={device_ip}"]
            if cmd_port:
                summary.append(f"cmd={cmd_port}/udp")
            if ts_port:
                summary.append(f"ts={ts_port}/udp")
            if dev_ver:
                summary.append(f"dev_ver={dev_ver}")
            if app_ver:
                summary.append(f"app_ver={app_ver}")
            print(f"  summary: {', '.join(summary)}")
            for key in sorted(headers):
                print(f"  {key}: {headers[key]}")
        print("  payload:")
        for line in printable(reply.payload).splitlines() or [""]:
            print(f"    {line}")
        # print(f"  hex: {reply.payload.hex()}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())