Andreas WredeandClaude Opus 4.8
ae3f2fc70f
CMD packets arrive as unauthenticated UDP datagrams, yet every hbc client executed the shell command they carry without any opt-in. Add an allow_remote_command config key, default false: when off, the command is logged and refused with "Refused: allow_remote_command is false" (visible in the server event log under the command service), and subprocess is never reached. When on, the client warns at startup that it will execute CMD packets. Applied to all four clients that handle CMD — hbc, hbc_windows.py, hbc_mini.py, and the C hbc_mini — since gating only one leaves the others wide open. The C client reads the same key from ~/.hbc.json and needs a rebuild to pick it up. UPD (self-update) is unchanged. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Tests for the allow_remote_command gate on CMD packets in hbc."""
|
|
import asyncio
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from hbd.client import main as client_main
|
|
from hbd.client.config import CLIENT_DEFAULTS, load_config
|
|
|
|
|
|
class FakeConn:
|
|
"""Stand-in for AsyncConnection that records what the client sends back."""
|
|
|
|
def __init__(self):
|
|
self.sent = []
|
|
|
|
async def sendto(self, msg, msg_id="HTB"):
|
|
self.sent.append(msg)
|
|
|
|
|
|
@pytest.fixture
|
|
def conn(monkeypatch):
|
|
"""A fake connection with remote commands disabled and subprocess booby-trapped."""
|
|
monkeypatch.setattr(client_main, "allow_remote_command", False)
|
|
|
|
def explode(*a, **kw):
|
|
raise AssertionError("subprocess must not run when the command is refused")
|
|
|
|
monkeypatch.setattr(subprocess, "check_output", explode)
|
|
return FakeConn()
|
|
|
|
|
|
def test_default_is_false():
|
|
assert CLIENT_DEFAULTS["allow_remote_command"] is False
|
|
assert load_config("/nonexistent/hbc.yaml")["allow_remote_command"] is False
|
|
assert client_main.allow_remote_command is False
|
|
|
|
|
|
def test_refused_when_disabled(conn):
|
|
asyncio.run(client_main.handle_command(conn, {"cmd": "id"}))
|
|
assert conn.sent == [{
|
|
"service": "command",
|
|
"msg": "Refused: allow_remote_command is false",
|
|
}]
|
|
|
|
|
|
def test_refusal_reported_for_every_attempt(conn):
|
|
for _ in range(3):
|
|
asyncio.run(client_main.handle_command(conn, {"cmd": "rm -rf /"}))
|
|
assert len(conn.sent) == 3
|
|
|
|
|
|
def test_empty_command_is_ignored_either_way(conn, monkeypatch):
|
|
asyncio.run(client_main.handle_command(conn, {}))
|
|
monkeypatch.setattr(client_main, "allow_remote_command", True)
|
|
asyncio.run(client_main.handle_command(conn, {"cmd": ""}))
|
|
assert conn.sent == []
|
|
|
|
|
|
def test_executed_when_enabled(conn, monkeypatch):
|
|
monkeypatch.setattr(client_main, "allow_remote_command", True)
|
|
monkeypatch.setattr(subprocess, "check_output", lambda *a, **kw: b"uid=0(root)")
|
|
asyncio.run(client_main.handle_command(conn, {"cmd": "id"}))
|
|
assert conn.sent == [{"service": "command", "msg": "OK uid=0(root)"}]
|
|
|
|
|
|
def test_failure_reported_when_enabled(conn, monkeypatch):
|
|
monkeypatch.setattr(client_main, "allow_remote_command", True)
|
|
|
|
def fail(*a, **kw):
|
|
raise subprocess.CalledProcessError(1, "false")
|
|
|
|
monkeypatch.setattr(subprocess, "check_output", fail)
|
|
asyncio.run(client_main.handle_command(conn, {"cmd": "false"}))
|
|
assert len(conn.sent) == 1
|
|
assert conn.sent[0]["msg"].startswith("CalledProcessError ")
|
|
|
|
|
|
def test_config_file_can_opt_in(tmp_path):
|
|
cfg_file = tmp_path / "hbc.yaml"
|
|
cfg_file.write_text("allow_remote_command: true\n", encoding="utf-8")
|
|
assert load_config(str(cfg_file))["allow_remote_command"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("configured, expected", [(True, True), (False, False), (None, False)])
|
|
def test_async_main_sets_the_flag_from_config(monkeypatch, configured, expected):
|
|
"""async_main must apply the config value before any CMD packet can arrive."""
|
|
monkeypatch.setattr(client_main, "allow_remote_command", not expected)
|
|
|
|
class Bail(Exception):
|
|
pass
|
|
|
|
# gethostname() is the first call after the flag is assigned — bail there so the
|
|
# test never opens a socket.
|
|
def bail():
|
|
raise Bail
|
|
|
|
monkeypatch.setattr(client_main.socket, "gethostname", bail)
|
|
|
|
cfg = {} if configured is None else {"allow_remote_command": configured}
|
|
with pytest.raises(Bail):
|
|
asyncio.run(client_main.async_main(object(), cfg))
|
|
assert client_main.allow_remote_command is expected
|