username/bouncer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: channel key support for +k channels

Browse Source 
Add channel_keys dict to NetworkConfig for storing per-channel keys.
Keys are used in KICK rejoin, passed via AUTOJOIN +#channel key syntax,
supported in ADDNETWORK channel_keys= parameter, and propagated through
REHASH. Extract rehash() as reusable async function for SIGHUP reuse.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-21 19:03:23 +01:00
parent bf4a589fc5
commit c11bd5555a
7 changed files with 240 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
tests/test_commands.py
View File

@@ -14,7 +14,8 @@ from bouncer.network import State
def _make_network(name: str, state: State, nick: str = "testnick",
host: str | None = None, channels: set[str] | None = None,
topics: dict[str, str] | None = None) -> MagicMock:
topics: dict[str, str] | None = None,
channel_keys: dict[str, str] | None = None) -> MagicMock:
"""Create a mock Network."""
net = MagicMock()
net.cfg.name = name
@@ -22,6 +23,7 @@ def _make_network(name: str, state: State, nick: str = "testnick",
net.cfg.port = 6697
net.cfg.tls = True
net.cfg.channels = list(channels) if channels else []
net.cfg.channel_keys = dict(channel_keys) if channel_keys else {}
net.cfg.nick = nick
net.cfg.password = None
net.state = state
@@ -514,6 +516,11 @@ class TestRehash:
old_net = _make_network("libera", State.READY)
router = _make_router(old_net)
router._notifier = MagicMock()
router._farm = MagicMock()
router._farm._cfg = BouncerConfig()
router._farm.start = AsyncMock()
router._farm.stop = AsyncMock()
new_cfg = Config(
bouncer=BouncerConfig(),
@@ -535,6 +542,101 @@ class TestRehash:
router.add_network.assert_awaited()
class TestRehashFunction:
@pytest.mark.asyncio
async def test_rehash_function_directly(self) -> None:
from bouncer.commands import rehash
from bouncer.config import BouncerConfig, Config, NetworkConfig, ProxyConfig
old_net = _make_network("libera", State.READY)
router = _make_router(old_net)
router._notifier = MagicMock()
router._farm = MagicMock()
router._farm._cfg = BouncerConfig()
router._farm.start = AsyncMock()
router._farm.stop = AsyncMock()
new_cfg = Config(
bouncer=BouncerConfig(),
proxy=ProxyConfig(),
networks={
"oftc": NetworkConfig(name="oftc", host="irc.oftc.net", port=6697, tls=True),
},
)
with patch("bouncer.config.load", return_value=new_cfg):
lines = await rehash(router, Path("/tmp/test.toml"))
assert lines[0] == "[REHASH]"
assert any("removed: libera" in line for line in lines)
assert any("added: oftc" in line for line in lines)
@pytest.mark.asyncio
async def test_rehash_updates_bouncer_config(self) -> None:
from bouncer.commands import rehash
from bouncer.config import BouncerConfig, Config, NetworkConfig, ProxyConfig
net = _make_network("libera", State.READY)
router = _make_router(net)
router._notifier = MagicMock()
router._farm = MagicMock()
router._farm._cfg = BouncerConfig()
router._farm.start = AsyncMock()
router._farm.stop = AsyncMock()
new_cfg = Config(
bouncer=BouncerConfig(notify_url="https://ntfy.sh/test"),
proxy=ProxyConfig(),
networks={
"libera": NetworkConfig(name="libera", host="irc.libera.chat",
port=6697, tls=True,
channel_keys={"#secret": "key"}),
},
)
with patch("bouncer.config.load", return_value=new_cfg):
result = await rehash(router, Path("/tmp/test.toml"))
assert result[0] == "[REHASH]"
assert router.config == new_cfg
# Notifier was replaced (new instance)
assert router._notifier is not None
@pytest.mark.asyncio
async def test_rehash_propagates_channel_keys(self) -> None:
from bouncer.commands import rehash
from bouncer.config import BouncerConfig, Config, NetworkConfig, ProxyConfig
net = _make_network("libera", State.READY)
net.cfg.host = "irc.libera.chat"
net.cfg.port = 6697
net.cfg.tls = True
net.cfg.proxy_host = None
net.cfg.proxy_port = None
net.cfg.channel_keys = {}
router = _make_router(net)
router._notifier = MagicMock()
router._farm = MagicMock()
router._farm._cfg = BouncerConfig()
router._farm.start = AsyncMock()
router._farm.stop = AsyncMock()
new_cfg = Config(
bouncer=BouncerConfig(),
proxy=ProxyConfig(),
networks={
"libera": NetworkConfig(name="libera", host="irc.libera.chat",
port=6697, tls=True,
channel_keys={"#secret": "key123"}),
},
)
with patch("bouncer.config.load", return_value=new_cfg):
await rehash(router, Path("/tmp/test.toml"))
assert net.cfg.channel_keys == {"#secret": "key123"}
class TestAddNetwork:
@pytest.mark.asyncio
async def test_addnetwork_missing_args(self) -> None:
@@ -648,6 +750,33 @@ class TestAutojoin:
lines = await commands.dispatch("AUTOJOIN libera -#missing", router, client)
assert any("not in autojoin" in line for line in lines)
@pytest.mark.asyncio
async def test_autojoin_with_key(self) -> None:
net = _make_network("libera", State.READY)
net.cfg.channels = []
net.cfg.channel_keys = {}
router = _make_router(net)
client = _make_client()
lines = await commands.dispatch("AUTOJOIN libera +#secret hunter2", router, client)
assert "[AUTOJOIN]" in lines[0]
assert any("added: #secret" in line for line in lines)
assert "#secret" in net.cfg.channels
assert net.cfg.channel_keys["#secret"] == "hunter2"
@pytest.mark.asyncio
async def test_autojoin_remove_clears_key(self) -> None:
net = _make_network("libera", State.READY,
channels={"#secret"},
channel_keys={"#secret": "hunter2"})
net.cfg.channels = ["#secret"]
net.cfg.channel_keys = {"#secret": "hunter2"}
router = _make_router(net)
client = _make_client()
lines = await commands.dispatch("AUTOJOIN libera -#secret", router, client)
assert any("removed: #secret" in line for line in lines)
assert "#secret" not in net.cfg.channels
assert "#secret" not in net.cfg.channel_keys
@pytest.mark.asyncio
async def test_autojoin_invalid_spec(self) -> None:
net = _make_network("libera", State.READY)

22
tests/test_config.py
View File

@@ -124,6 +124,28 @@ tls = true
cfg = load(_write_config(config))
assert cfg.networks["test"].port == 6697
def test_channel_keys_parsed(self):
config = """\
[bouncer]
password = "x"
[proxy]
[networks.test]
host = "irc.example.com"
channels = ["#secret", "#public"]
channel_keys = { "#secret" = "hunter2" }
"""
cfg = load(_write_config(config))
net = cfg.networks["test"]
assert net.channel_keys == {"#secret": "hunter2"}
assert "#secret" in net.channels
def test_channel_keys_default_empty(self):
cfg = load(_write_config(MINIMAL_CONFIG))
net = cfg.networks["test"]
assert net.channel_keys == {}
def test_operational_defaults(self):
"""Ensure all operational values have sane defaults."""
cfg = load(_write_config(MINIMAL_CONFIG))

19
tests/test_network.py
View File

@@ -479,6 +479,25 @@ class TestHandleKick:
# Should rejoin (rejoin_delay=0)
writer.write.assert_called_with(b"JOIN #test\r\n")
@pytest.mark.asyncio
async def test_kick_rejoin_with_key(self) -> None:
cfg = _cfg(channels=["#secret"])
cfg.channel_keys = {"#secret": "hunter2"}
net = _net(cfg=cfg)
net.state = State.READY
net._running = True
net.nick = "me"
net.channels = {"#secret"}
writer = MagicMock()
writer.is_closing.return_value = False
writer.drain = AsyncMock()
net._writer = writer
await net._handle(_msg(":op!user@host KICK #secret me :reason"))
assert "#secret" not in net.channels
# Should rejoin with key (rejoin_delay=0)
writer.write.assert_called_with(b"JOIN #secret hunter2\r\n")
@pytest.mark.asyncio
async def test_kick_other_user_ignored(self) -> None:
net = _net()