daniel156161
541b950519
Both tables grew one entry per distinct key and never shrank, so a long-running SDK client or a public serve process accumulated state for every endpoint or identity it had ever seen. GC cannot reclaim them while the pool and limiter still reference them. Cap the pool at 64 endpoint buckets and the limiter at 4096 buckets. Both evict useless state first: connections past the idle timeout the server has likely dropped anyway, and buckets that have fully refilled, which carry no throttling information. Only then fall back to evicting the oldest entry. Evicting a limiter bucket resets throttling for that identity, which is the deliberate trade: an attacker cycling identities faster than they go idle can regain tokens, but unbounded growth would take the process down instead.
226 lines
10 KiB
Python
226 lines
10 KiB
Python
"""Unit tests for serve-side security: per-key policy loading, rate limiting, context."""
|
|
import pytest
|
|
|
|
from browser_cli.auth.keys import (
|
|
_parse_authorized_line,
|
|
format_authorized_line,
|
|
load_authorized_keys_with_names,
|
|
load_authorized_keys_with_policies,
|
|
set_authorized_key_policy,
|
|
)
|
|
from browser_cli.command_security import CommandPolicy, assert_command_allowed
|
|
from browser_cli.serve.security import (
|
|
RateLimiter,
|
|
ServeSecurity,
|
|
key_policies_from_authorized_keys,
|
|
policy_from_categories,
|
|
)
|
|
|
|
# ── policy_from_categories ───────────────────────────────────────────────────────
|
|
|
|
def test_policy_from_categories_all_is_unrestricted():
|
|
assert policy_from_categories(["all"]) == CommandPolicy.unrestricted()
|
|
|
|
def test_policy_from_categories_subset():
|
|
policy = policy_from_categories(["read-page", "control"])
|
|
assert policy == CommandPolicy(allow_read_page=True, allow_control=True)
|
|
assert policy.allow_dangerous is False
|
|
|
|
def test_policy_from_categories_safe_and_empty_are_noops():
|
|
assert policy_from_categories(["safe"]) == CommandPolicy()
|
|
assert policy_from_categories([]) == CommandPolicy()
|
|
|
|
def test_policy_from_categories_rejects_unknown():
|
|
with pytest.raises(ValueError, match="unknown command category"):
|
|
policy_from_categories(["bogus"])
|
|
|
|
def test_policy_from_categories_keys():
|
|
assert policy_from_categories(["keys"]) == CommandPolicy(allow_keys=True)
|
|
|
|
# ── keys category gating ─────────────────────────────────────────────────────────
|
|
|
|
def test_key_commands_are_keys_category():
|
|
from browser_cli.command_security import command_category
|
|
assert command_category("browser-cli.auth.keys") == "keys"
|
|
assert command_category("browser-cli.auth.trust") == "keys"
|
|
assert command_category("browser-cli.auth.policy") == "keys"
|
|
assert command_category("browser-cli.targets") == "safe" # discovery stays open
|
|
|
|
def test_key_commands_blocked_without_allow_keys():
|
|
for cmd in ("browser-cli.auth.keys", "browser-cli.auth.trust", "browser-cli.auth.policy"):
|
|
with pytest.raises(PermissionError):
|
|
assert_command_allowed(cmd, CommandPolicy()) # safe-only default
|
|
assert_command_allowed(cmd, CommandPolicy(allow_keys=True)) # explicit grant
|
|
assert_command_allowed(cmd, CommandPolicy.unrestricted()) # all includes keys
|
|
|
|
def test_full_control_still_cannot_manage_keys():
|
|
"""A key with control+dangerous (but not keys) cannot list/trust keys."""
|
|
policy = CommandPolicy(allow_read_page=True, allow_control=True, allow_dangerous=True)
|
|
with pytest.raises(PermissionError):
|
|
assert_command_allowed("browser-cli.auth.trust", policy)
|
|
|
|
# ── set_authorized_key_policy ────────────────────────────────────────────────────
|
|
|
|
def test_set_policy_updates_by_pubkey(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
pub = "a" * 64
|
|
path.write_text(f"{pub} laptop\n")
|
|
assert set_authorized_key_policy(path, pub, ["control"]) == (pub, "laptop")
|
|
assert load_authorized_keys_with_policies(path) == [(pub, "laptop", ["control"])]
|
|
|
|
def test_set_policy_by_name_and_remove_with_none(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
pub = "b" * 64
|
|
path.write_text(f"{pub} ci-bot allow:all\n")
|
|
assert set_authorized_key_policy(path, "ci-bot", None) == (pub, "ci-bot") # remove token
|
|
assert load_authorized_keys_with_policies(path) == [(pub, "ci-bot", None)]
|
|
|
|
def test_set_policy_safe_only_writes_empty_token(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
pub = "c" * 64
|
|
path.write_text(f"{pub} reader\n")
|
|
set_authorized_key_policy(path, pub, [])
|
|
assert path.read_text().strip() == f"{pub} reader allow:"
|
|
|
|
def test_set_policy_not_found_returns_none(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
path.write_text(f"{'a' * 64} laptop\n")
|
|
assert set_authorized_key_policy(path, "nonexistent", ["control"]) is None
|
|
|
|
def test_set_policy_ambiguous_name_raises(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
path.write_text(f"{'a' * 64} dup\n{'b' * 64} dup\n")
|
|
with pytest.raises(ValueError, match="ambiguous"):
|
|
set_authorized_key_policy(path, "dup", ["control"])
|
|
|
|
def test_set_policy_preserves_other_lines(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
a, b = "a" * 64, "b" * 64
|
|
path.write_text(f"{a} first\n{b} second allow:read-page\n")
|
|
set_authorized_key_policy(path, a, ["control"])
|
|
assert load_authorized_keys_with_policies(path) == [
|
|
(a, "first", ["control"]),
|
|
(b, "second", ["read-page"]), # untouched
|
|
]
|
|
|
|
# ── authorized_keys line parsing ─────────────────────────────────────────────────
|
|
|
|
def test_parse_line_pubkey_only():
|
|
assert _parse_authorized_line("abc123") == ("abc123", "", None)
|
|
|
|
def test_parse_line_name_with_spaces_no_policy():
|
|
# A multi-word name (e.g. "YubiKey 5C NFC FIPS") must stay intact, policy None.
|
|
assert _parse_authorized_line("abc YubiKey 5C NFC FIPS") == ("abc", "YubiKey 5C NFC FIPS", None)
|
|
|
|
def test_parse_line_name_with_spaces_and_policy():
|
|
pub, name, cats = _parse_authorized_line("abc YubiKey 5C NFC FIPS allow:read-page,control")
|
|
assert pub == "abc"
|
|
assert name == "YubiKey 5C NFC FIPS" # allow: token stripped out of the name
|
|
assert cats == ["read-page", "control"]
|
|
|
|
def test_parse_line_empty_allow_is_safe():
|
|
assert _parse_authorized_line("abc name allow:") == ("abc", "name", [])
|
|
|
|
def test_parse_line_skips_comments_and_blanks():
|
|
assert _parse_authorized_line("# comment") is None
|
|
assert _parse_authorized_line(" ") is None
|
|
|
|
def test_format_authorized_line_roundtrips():
|
|
line = format_authorized_line("abc", "my laptop", ["read-page", "control"])
|
|
assert line == "abc my laptop allow:read-page,control"
|
|
assert _parse_authorized_line(line) == ("abc", "my laptop", ["read-page", "control"])
|
|
# No categories → no allow token.
|
|
assert format_authorized_line("abc", "laptop") == "abc laptop"
|
|
|
|
# ── key_policies_from_authorized_keys ────────────────────────────────────────────
|
|
|
|
def test_key_policies_from_authorized_keys(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
path.write_text(
|
|
"AABBCC laptop allow:all\n"
|
|
"ddee01 ci-bot allow:read-page,control\n"
|
|
"112233 readonly\n" # no allow token → no override entry
|
|
)
|
|
policies = key_policies_from_authorized_keys(path)
|
|
assert policies["aabbcc"] == CommandPolicy.unrestricted() # normalised to lowercase
|
|
assert policies["ddee01"] == CommandPolicy(allow_read_page=True, allow_control=True)
|
|
assert "112233" not in policies # falls back to server default
|
|
|
|
def test_key_policies_none_returns_empty():
|
|
assert key_policies_from_authorized_keys(None) == {}
|
|
|
|
def test_key_policies_rejects_unknown_category(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
path.write_text("abc name allow:bogus\n")
|
|
with pytest.raises(ValueError, match="unknown command category"):
|
|
key_policies_from_authorized_keys(path)
|
|
|
|
def test_load_with_names_ignores_allow_token(tmp_path):
|
|
path = tmp_path / "authorized_keys"
|
|
path.write_text("abc my laptop allow:control\n")
|
|
assert load_authorized_keys_with_names(path) == [("abc", "my laptop")]
|
|
assert load_authorized_keys_with_policies(path) == [("abc", "my laptop", ["control"])]
|
|
|
|
# ── RateLimiter ──────────────────────────────────────────────────────────────────
|
|
|
|
def test_rate_limiter_zero_rate_never_limits():
|
|
limiter = RateLimiter(rate=0)
|
|
assert all(limiter.allow("k") for _ in range(1000))
|
|
|
|
def test_rate_limiter_burst_then_block():
|
|
limiter = RateLimiter(rate=0.0001, burst=3)
|
|
assert limiter.allow("k") is True
|
|
assert limiter.allow("k") is True
|
|
assert limiter.allow("k") is True
|
|
assert limiter.allow("k") is False # bucket drained, refill negligible
|
|
|
|
def test_rate_limiter_is_per_key():
|
|
limiter = RateLimiter(rate=0.0001, burst=1)
|
|
assert limiter.allow("a") is True
|
|
assert limiter.allow("b") is True # different key has its own bucket
|
|
assert limiter.allow("a") is False
|
|
assert limiter.allow("b") is False
|
|
|
|
def test_rate_limiter_caps_identity_buckets():
|
|
limiter = RateLimiter(rate=0.0001, burst=1, max_buckets=3)
|
|
for i in range(10):
|
|
assert limiter.allow(f"key-{i}") is True
|
|
assert len(limiter._buckets) <= 3
|
|
|
|
def test_rate_limiter_prunes_refilled_idle_buckets(monkeypatch):
|
|
current = 1000.0
|
|
monkeypatch.setattr("browser_cli.serve.security.time.monotonic", lambda: current)
|
|
limiter = RateLimiter(rate=1, burst=2, max_buckets=2)
|
|
assert limiter.allow("old") is True
|
|
current += 120.0
|
|
assert limiter.allow("a") is True
|
|
assert limiter.allow("b") is True
|
|
assert "old" not in limiter._buckets
|
|
assert len(limiter._buckets) <= 2
|
|
|
|
# ── ServeSecurity ────────────────────────────────────────────────────────────────
|
|
|
|
def test_effective_policy_prefers_per_key_override():
|
|
sec = ServeSecurity(
|
|
policy=CommandPolicy.unrestricted(),
|
|
key_policies={"abc": CommandPolicy()},
|
|
)
|
|
assert sec.effective_policy("abc") == CommandPolicy() # override
|
|
assert sec.effective_policy("other") == CommandPolicy.unrestricted() # default
|
|
assert sec.effective_policy(None) == CommandPolicy.unrestricted()
|
|
# And the override actually gates a dangerous command:
|
|
with pytest.raises(PermissionError):
|
|
assert_command_allowed("dom.eval", sec.effective_policy("abc"))
|
|
|
|
def test_label_for_renders_name_and_short_pubkey():
|
|
sec = ServeSecurity(key_names={"ab12cd34ef": "laptop"})
|
|
assert sec.label_for("ab12cd34ef") == "laptop ab12cd34…"
|
|
assert sec.label_for("ffeeddccbb") == "ffeeddcc…" # unknown key → short pubkey only
|
|
assert sec.label_for(None) is None
|
|
|
|
def test_serve_security_defaults_are_safe():
|
|
sec = ServeSecurity()
|
|
assert sec.key_policies == {}
|
|
assert sec.key_names == {}
|
|
assert sec.rate_limiter is None
|