diff --git a/tests/test_app.py b/tests/test_app.py index 82356fd..a8d4645 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -54,6 +54,11 @@ def test_ui_serves_self_contained_operator_console(): assert 'id="operations-grid"' in response.text assert 'id="doctor-form"' in response.text assert 'id="preflight-form"' in response.text + assert 'id="sharing-form"' in response.text + assert 'id="sharing-events"' in response.text + assert "/emby/account-sharing" in response.text + assert 'id="login"' not in response.text + assert "Butler-Token" not in response.text assert "localStorage" not in response.text @@ -86,6 +91,131 @@ def test_ui_session_login_uses_httponly_cookie_and_csrf(): assert allowed.status_code == 200 +def test_ui_anonymous_session_is_automatic_and_strictly_read_only(): + app._ui_sessions.clear() + with TestClient(app.app) as client: + session = client.get("/ui/session") + assert session.status_code == 200 + assert session.json()["authenticated"] is True + assert session.json()["read_only"] is True + assert "HttpOnly" in session.headers.get("set-cookie", "") + + capabilities = client.get("/capabilities") + assert capabilities.status_code == 200 + + csrf = client.cookies.get("butler_csrf") + mutation = client.post("/config/reload", headers={"X-CSRF-Token": csrf}) + assert mutation.status_code == 403 + assert "read-only" in mutation.json()["detail"].lower() + + +def test_emby_network_identity_normalizes_ipv4_and_ipv6_privacy_addresses(): + ipv4 = app._emby_network_identity("203.0.113.9:443") + assert ipv4["ip"] == "203.0.113.9" + assert ipv4["network"] == "203.0.113.9/32" + assert ipv4["identity"] == "203.0.113.9/32" + + first = app._emby_network_identity("2003:abcd:1234:5678::1") + privacy_peer = app._emby_network_identity("[2003:abcd:1234:5678:ffff::99]:443") + sibling_subnet = app._emby_network_identity("2003:abcd:1234:9999::1") + assert first["network"] == privacy_peer["network"] == "2003:abcd:1234:5678::/64" + assert first["parent"] == sibling_subnet["parent"] == "2003:abcd:1234::/48" + assert first["identity"] == sibling_subnet["identity"] == "2003:abcd:1234::/48" + + +def test_emby_sharing_flags_concurrent_distinct_networks_but_not_sibling_ipv6_subnets(): + def series(endpoint, values, city): + return { + "metric": { + "job": "emby-sascha", "username": "Alice", "remoteEndPoint": endpoint, + "city": city, "region": "Test", "countryCode": "DE", + "latitude": "48.1", "longitude": "11.5", + }, + "values": [[timestamp, "1"] for timestamp in values], + } + + payload = [ + series("2606:4700:1234:1000::1", [100, 160, 220, 280, 340, 400], "Home"), + series("2606:4700:1234:2000::2", [100, 160, 220, 280, 340, 400], "Home privacy subnet"), + series("2001:4860:9999:1000::1", [100, 160, 220, 280, 340, 400], "Away"), + ] + + result = app._analyze_emby_sharing(payload, step_seconds=60) + + assert result["summary"]["concurrent_events"] == 1 + event = result["events"][0] + assert event["type"] == "concurrent_networks" + assert event["username"] == "Alice" + assert len(event["evidence"]) == 2 + assert {item["identity"] for item in event["evidence"]} == { + "2606:4700:1234::/48", "2001:4860:9999::/48" + } + + +def test_emby_sharing_ignores_short_overlap_inside_prometheus_staleness_window(): + def series(endpoint): + return {"metric": {"job": "emby-sascha", "username": "Alice", "remoteEndPoint": endpoint, + "city": "Munich", "region": "Bavaria", "countryCode": "DE", + "latitude": "48.1", "longitude": "11.5"}, + "values": [[100, "1"], [160, "1"]]} + + result = app._analyze_emby_sharing([series("8.8.8.8"), series("1.1.1.1")], step_seconds=60) + + assert result["summary"]["concurrent_events"] == 0 + + +def test_emby_sharing_flags_geographically_impossible_network_change(): + payload = [ + { + "metric": {"job": "emby-chris", "username": "Bob", "remoteEndPoint": "8.8.8.8", + "city": "Berlin", "region": "Berlin", "countryCode": "DE", + "latitude": "52.5200", "longitude": "13.4050"}, + "values": [[100, "1"], [160, "1"]], + }, + { + "metric": {"job": "emby-chris", "username": "Bob", "remoteEndPoint": "1.1.1.1", + "city": "New York", "region": "New York", "countryCode": "US", + "latitude": "40.7128", "longitude": "-74.0060"}, + "values": [[400, "1"], [460, "1"]], + }, + ] + + result = app._analyze_emby_sharing(payload, step_seconds=60) + + assert result["summary"]["concurrent_events"] == 0 + assert result["summary"]["impossible_travel_events"] == 1 + event = result["events"][0] + assert event["type"] == "impossible_travel" + assert event["distance_km"] > 6000 + assert event["required_speed_kmh"] > 1000 + + +def test_emby_account_sharing_endpoint_is_read_only_and_filterable(monkeypatch): + async def history(days, server): + assert days == 7 + assert server == "all" + return ([{ + "metric": {"job": "emby-sascha", "username": "Alice", "remoteEndPoint": "8.8.8.8", + "city": "Munich", "region": "Bavaria", "countryCode": "DE", + "latitude": "48.1", "longitude": "11.5"}, + "values": [[100, "1"], [160, "1"]], + }], 60) + + monkeypatch.setattr(app, "_fetch_emby_session_history", history) + with TestClient(app.app) as client: + response = client.get( + "/emby/account-sharing?days=7&username=alice", + headers={"Authorization": "Bearer test-token"}, + ) + + assert response.status_code == 200 + payload = response.json() + assert payload["policy"]["mode"] == "conservative" + assert payload["policy"]["ipv6_detection_identity"] == "/48" + assert payload["summary"]["users_analyzed"] == 1 + assert payload["users"][0]["username"] == "Alice" + + def test_capabilities_is_live_machine_readable_safety_map(): with TestClient(app.app) as client: response = client.get(