Files
cc-ci/tests/rustdesk-server/custom/test_punch_hole_no_amplification.py
T
autonomic-bot f5f11cd769
continuous-integration/drone/push Build is passing
chore(rustdesk): drop unused import
2026-10-30 19:03:11 +00:00

90 lines
4.4 KiB
Python

"""rustdesk-server — security regression: the unauthenticated punch-hole amplification fix.
Upstream 1.1.16 fixed GH rustdesk/rustdesk-server#670 (reserved-then-withdrawn
CVE-2026-30784): hbbs used to answer ANY UDP punch-hole request with a punch-hole response —
no check the requester was a registered peer or that the exchange was solicited — so spoofed
source IPs turned public servers into DDoS reflection/amplifiers. This happened EVEN with
`-k _` enforced (which this recipe uses), because the key gate never covered the punch-hole
path.
Empirical baseline (cc-ci dev deploy, 2026-10-30, hand-encoded RendezvousMessage protobufs):
1.1.12: register_peer -> 3a021001 (register_peer_response, request_pk=true)
punch_hole_request x3 -> 5a021803 EVERY time (punch_hole_response, failure=3)
1.1.16: register_peer -> 3a021001 (unchanged — the channel still answers)
punch_hole_request x3 -> SILENCE (the fix)
The register_peer sanity check inside the same test is what makes "silence" meaningful: it
proves the server is up, our encoding is right, and UDP 21116 is answering — so a silent
punch-hole exchange can only be the FIX, not a broken server or a bad probe. Non-vacuous: a
regression of the fix makes hbbs answer again and this test goes RED naming the layer.
Version-gated (STYLE.md §2): the fixed behavior only exists from 1.1.16; older versions are
exercised by the upgrade tier's base deploy and are EXPECTED to answer.
"""
from __future__ import annotations
import os
import sys
import pytest
sys.path.insert(0, os.path.dirname(__file__))
import _rustdesk_proto as proto # noqa: E402
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "runner"))
def _deployed_app_version(domain: str) -> tuple[int, int, int]:
"""The running app service's image tag (e.g. rustdesk/rustdesk-server:1.1.16 -> (1,1,16)),
via the harness's first-class deployed_identity (labels + image in one inspect). Fails
LOUDLY on an unparsable image rather than returning None — a gate that cannot decide must
never silently run (vacuous-green) or silently skip (the first enrollment run shipped a
None-returning hand-rolled inspect and observed pre-fix answering on a 1.1.12 head)."""
from harness import lifecycle # noqa: PLC0415 (lazy: keeps import cost off collection)
image = (lifecycle.deployed_identity(domain).get("image") or "").strip()
tag = image.split("@", 1)[0].rsplit(":", 1)[-1]
try:
return tuple(int(x) for x in tag.split(".")[:3]) # type: ignore[return-value]
except ValueError:
pytest.fail(
f"cannot determine deployed rustdesk-server version from image {image!r} — fix "
"the read before trusting this test's verdict"
)
def test_unauthenticated_punch_hole_is_not_answered(live_app):
version = _deployed_app_version(live_app)
if version < (1, 1, 16):
pytest.skip(
f"amplification fix landed upstream in 1.1.16 (running {version}) — pre-fix "
"answering is expected behavior, covered by the upgrade tier's base deploy"
)
# 1) channel sanity: a plain register_peer MUST be answered (register_peer_response,
# RendezvousMessage field 7 -> first varint key 0x3A). Without this, later silence
# could mean "server down / encoding wrong", not "fixed".
reg = proto.retry_register_peer("ccci-sanity-probe")
assert reg is not None, (
"register_peer sanity probe unanswered — hbbs UDP 21116 not serving; cannot judge "
"the punch-hole path"
)
assert proto.parse_union_field(reg) == 7, (
f"expected register_peer_response (oneof field 7), got {proto.parse_union_field(reg)}: {reg[:40].hex()}"
)
# 2) the actual security property: an UNAUTHENTICATED punch-hole request (no
# licence_key/token, spoofable shape) for a nonexistent target must get NO response.
# Pre-fix (1.1.12) every request was answered with punch_hole_response (field 11) —
# the reflection/amplification vector. 3 independent attempts, all silent.
answered = []
for attempt in range(3):
resp = proto.udp_probe(proto.punch_hole_request(f"ccci-no-such-peer-{attempt}"))
if resp is not None:
answered.append((attempt, resp[:40].hex()))
assert not answered, (
"hbbs answered unauthenticated punch-hole request(s) — the 1.1.16 amplification fix "
f"(GH #670) has REGRESSED: {answered}"
)