homelab/tests/radius/mock_udm.py

"""Stand-in for the UDM RADIUS (UDP 1812) that UniFi Endpoint logins go to.

Terminates nothing; logs what arrives and answers just enough EAP to prove
that FreeRADIUS proxies the whole conversation:
  round 1 (EAP-Response/Identity, no State)
    identity endpoint-reject  Access-Reject + EAP-Failure
    any other identity        Access-Challenge + State + EAP-Request/TTLS-Start
  round 2 (State from round 1)
    Access-Accept + EAP-Success, Tunnel-* VLAN 200, MS-MPPE-Recv-Key 0x11 * 32,
    MS-MPPE-Send-Key 0x22 * 32
  no EAP-Message (status probe), unknown State  Access-Reject
All replies carry Message-Authenticator and echo Proxy-State.
"""

import hashlib
import os
import socket
import struct

from mock_outpost import ATTR_USER_NAME, attr, build_reply, parse_attrs, request_ma_ok, vsa

SECRET = os.environ["MOCK_SECRET"].encode()

ACCESS_REQUEST, ACCESS_ACCEPT, ACCESS_REJECT, ACCESS_CHALLENGE = 1, 2, 3, 11
ATTR_STATE, ATTR_PROXY_STATE, ATTR_EAP_MESSAGE = 24, 33, 79
EAP_REQUEST, EAP_RESPONSE, EAP_SUCCESS, EAP_FAILURE = 1, 2, 3, 4
EAP_IDENTITY, EAP_TTLS = 1, 21

STATE = b"udm-mock-state-1"
RECV_KEY = b"\x11" * 32
SEND_KEY = b"\x22" * 32
TUNNEL_VLAN_200 = (
    attr(64, b"\x00\x00\x00\x0d")  # Tunnel-Type = VLAN
    + attr(65, b"\x00\x00\x00\x06")  # Tunnel-Medium-Type = IEEE-802
    + attr(81, b"200")  # Tunnel-Private-Group-Id = "200"
)


def mppe_key(key, req_auth, salt):
    """RFC 2548 2.4.2 encryption of an MS-MPPE key."""
    plain = bytes([len(key)]) + key
    plain += b"\x00" * (-len(plain) % 16)
    out, last = b"", req_auth + salt
    for i in range(0, len(plain), 16):
        b = hashlib.md5(SECRET + last).digest()
        chunk = bytes(p ^ k for p, k in zip(plain[i : i + 16], b))
        out += chunk
        last = chunk
    return salt + out


def eap(code, ident, payload=b""):
    return struct.pack("!BBH", code, ident, 4 + len(payload)) + payload


def main():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(("0.0.0.0", 1812))
    print("mock udm listening on udp/1812", flush=True)
    while True:
        packet, peer = sock.recvfrom(4096)
        if len(packet) < 20:
            continue
        code, ident, _length = struct.unpack("!BBH", packet[:4])
        req_auth = packet[4:20]
        if code != ACCESS_REQUEST:
            print(f"peer={peer[0]} code={code} ignored", flush=True)
            continue
        attrs = parse_attrs(packet[20:])
        user = next((v for t, _, v in attrs if t == ATTR_USER_NAME), b"").decode(errors="replace")
        state = next((v for t, _, v in attrs if t == ATTR_STATE), None)
        eap_msg = b"".join(v for t, _, v in attrs if t == ATTR_EAP_MESSAGE)
        echo = b"".join(attr(ATTR_PROXY_STATE, v) for t, _, v in attrs if t == ATTR_PROXY_STATE)
        ma = request_ma_ok(packet, attrs)

        if len(eap_msg) < 5:
            reply = build_reply(ACCESS_REJECT, ident, req_auth, echo, True)
            print(f"peer={peer[0]} probe user={user} request_ma={ma} -> reject", flush=True)
        elif state is None:
            eap_id = eap_msg[1]
            eap_ok = eap_msg[0] == EAP_RESPONSE and eap_msg[4] == EAP_IDENTITY
            identity = eap_msg[5:].decode(errors="replace")
            if identity == "endpoint-reject":
                body = attr(ATTR_EAP_MESSAGE, eap(EAP_FAILURE, eap_id))
                reply = build_reply(ACCESS_REJECT, ident, req_auth, echo + body, True)
                result = "reject"
            else:
                body = attr(ATTR_STATE, STATE) + attr(
                    ATTR_EAP_MESSAGE, eap(EAP_REQUEST, (eap_id + 1) & 0xFF, bytes([EAP_TTLS, 0x20]))
                )
                reply = build_reply(ACCESS_CHALLENGE, ident, req_auth, echo + body, True)
                result = "challenge"
            print(
                f"peer={peer[0]} round=1 user={user} eap_identity_response={eap_ok} "
                f"eap_identity={identity} request_ma={ma} -> {result}",
                flush=True,
            )
        elif state == STATE:
            eap_id = eap_msg[1]
            eap_type = eap_msg[4]
            body = (
                TUNNEL_VLAN_200
                + vsa(311, 17, mppe_key(RECV_KEY, req_auth, b"\x80\x01"))
                + vsa(311, 16, mppe_key(SEND_KEY, req_auth, b"\x80\x02"))
                + attr(ATTR_EAP_MESSAGE, eap(EAP_SUCCESS, eap_id))
            )
            reply = build_reply(ACCESS_ACCEPT, ident, req_auth, echo + body, True)
            print(
                f"peer={peer[0]} round=2 user={user} state_ok=True eap_type={eap_type} "
                f"request_ma={ma} -> accept",
                flush=True,
            )
        else:
            reply = build_reply(ACCESS_REJECT, ident, req_auth, echo, True)
            print(f"peer={peer[0]} user={user} state={state.hex()} unknown -> reject", flush=True)
        sock.sendto(reply, peer)


if __name__ == "__main__":
    main()