Add Emby account-sharing analysis and read-only UI access (tests/test_app.py)

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(