443 lines
15 KiB
Python
443 lines
15 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from adcloud_server.database import Database, identifiers
|
|
from adcloud_server.identity_models import StitchRequest
|
|
from adcloud_server.identity_service import IdentityService
|
|
from adcloud_server.main import create_app
|
|
from adcloud_server.settings import Settings
|
|
|
|
SECRET = "identity-test-secret-" + "x" * 32
|
|
SERVICE_TOKEN = "trusted-test-token-" + "y" * 32
|
|
AUTH = {"Authorization": "Bearer " + SERVICE_TOKEN}
|
|
|
|
|
|
@pytest.fixture
|
|
def settings(tmp_path):
|
|
return Settings(
|
|
database_url="sqlite:///" + str(tmp_path / "identity.db"),
|
|
identity_secret=SECRET,
|
|
service_token=SERVICE_TOKEN,
|
|
cookie_secure=False,
|
|
event_registry_required=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(settings):
|
|
with TestClient(create_app(settings)) as test_client:
|
|
yield test_client
|
|
|
|
|
|
def anonymous(client):
|
|
response = client.post("/v1/identity/anonymous")
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def stitch(client, identity, token=None, **kwargs):
|
|
body = {"identifiers": identity, **kwargs}
|
|
if token:
|
|
body["anonymous_token"] = token
|
|
response = client.post("/v1/identity/stitch", json=body, headers=AUTH)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def email(value="alice@example.com"):
|
|
return {"kind": "email", "value": value}
|
|
|
|
|
|
def profile(client, master):
|
|
response = client.get("/v1/users/" + master, headers=AUTH)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def history(client, master):
|
|
response = client.get("/v1/events", params={"master_id": master}, headers=AUTH)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def test_first_event_seeds_persistent_cookie_and_reuses_it(client):
|
|
first = client.post(
|
|
"/v1/events", json={"name": "page.view", "properties": {"path": "/shoes"}}
|
|
)
|
|
assert first.status_code == 200
|
|
assert "HttpOnly" in first.headers["set-cookie"]
|
|
assert "Max-Age=31536000" in first.headers["set-cookie"]
|
|
assert "SameSite=lax" in first.headers["set-cookie"]
|
|
assert first.headers["cache-control"] == "no-store"
|
|
second = anonymous(client)
|
|
assert first.json()["anonymous_id"] == second["anonymous_id"]
|
|
assert profile(client, second["master_id"])["event_count"] == 1
|
|
|
|
|
|
def test_cross_device_stitch_inherits_events_sparse_traits_and_tags(client):
|
|
known = anonymous(client)
|
|
stitch(client, [email()], attributes={"preferred_color": "blue"}, tags=["member"])
|
|
client.post("/v1/identity/anonymous?rotate=true")
|
|
visitor = anonymous(client)
|
|
client.patch(
|
|
"/v1/users/" + visitor["master_id"],
|
|
headers=AUTH,
|
|
json={
|
|
"attributes": {
|
|
"interest": "running",
|
|
"missing": None,
|
|
"nested": {"present": 1, "empty": None},
|
|
"preferred_color": "red",
|
|
},
|
|
"tags": ["runner"],
|
|
},
|
|
)
|
|
event = client.post(
|
|
"/v1/events", json={"name": "product.view", "properties": {"sku": "shoe-1"}}
|
|
).json()
|
|
result = stitch(client, [email()])
|
|
assert result["master_id"] == known["master_id"]
|
|
assert visitor["master_id"] in result["merged_master_ids"]
|
|
assert result["rotated_anonymous_id"] is False
|
|
merged = profile(client, result["master_id"])
|
|
assert merged["attributes"] == {
|
|
"preferred_color": "blue",
|
|
"interest": "running",
|
|
"nested": {"present": 1},
|
|
}
|
|
assert merged["tags"] == ["member", "runner"]
|
|
assert merged["event_count"] == 1
|
|
assert history(client, visitor["master_id"])["items"][0]["id"] == event["event_id"]
|
|
graph = client.get(
|
|
"/v1/identity/" + result["master_id"] + "/graph", headers=AUTH
|
|
).json()
|
|
assert len(graph["member_ids"]) == 2
|
|
assert len(graph["edges"]) == 1
|
|
assert graph["identifiers"][0]["display"] == "a***@example.com"
|
|
|
|
|
|
def test_transitive_verified_bridge_resolves_every_alias(client):
|
|
anonymous(client)
|
|
a = stitch(client, [email()])
|
|
client.post("/v1/identity/anonymous?rotate=true")
|
|
phone = {"kind": "phone", "value": "+1 (415) 555-2671"}
|
|
b = stitch(client, [phone])
|
|
client.post("/v1/identity/anonymous?rotate=true")
|
|
oauth = {
|
|
"kind": "oauth",
|
|
"namespace": "https://issuer.example/client-a",
|
|
"value": "subject-123",
|
|
}
|
|
c = stitch(client, [oauth])
|
|
joined = stitch(client, [email(), phone, oauth])
|
|
assert joined["master_id"] == a["master_id"]
|
|
for old in [a, b, c]:
|
|
assert profile(client, old["master_id"])["master_id"] == a["master_id"]
|
|
resolved = client.post(
|
|
"/v1/identity/resolve",
|
|
headers=AUTH,
|
|
json={
|
|
"identifiers": [email(), {"kind": "phone", "value": "+14155552671"}, oauth]
|
|
},
|
|
).json()
|
|
assert resolved == {
|
|
"master_id": a["master_id"],
|
|
"matches": [a["master_id"]],
|
|
"ambiguous": False,
|
|
}
|
|
|
|
|
|
def test_shared_browser_switches_accounts_without_merging_them(client):
|
|
a = stitch_after_anonymous(client, email())
|
|
client.post("/v1/events", json={"name": "alice.view"})
|
|
b = stitch(client, [email("bob@example.com")])
|
|
assert b["rotated_anonymous_id"] is True
|
|
assert a["master_id"] != b["master_id"]
|
|
assert profile(client, b["master_id"])["event_count"] == 0
|
|
assert profile(client, a["master_id"])["event_count"] == 1
|
|
|
|
|
|
def stitch_after_anonymous(client, identifier):
|
|
anonymous(client)
|
|
return stitch(client, [identifier])
|
|
|
|
|
|
def test_authenticated_backend_can_add_new_identity_to_current_account(client):
|
|
a = stitch_after_anonymous(client, email())
|
|
result = stitch(
|
|
client,
|
|
[{"kind": "phone", "value": "+14155552671"}],
|
|
account_master_id=a["master_id"],
|
|
)
|
|
assert result["master_id"] == a["master_id"]
|
|
assert result["rotated_anonymous_id"] is False
|
|
assert (
|
|
len(
|
|
client.get(
|
|
"/v1/identity/" + a["master_id"] + "/graph", headers=AUTH
|
|
).json()["identifiers"]
|
|
)
|
|
== 2
|
|
)
|
|
|
|
|
|
def test_namespace_isolates_third_party_subject_ids(client):
|
|
first = {"kind": "oauth", "namespace": "issuer/client-a", "value": "same-subject"}
|
|
second = {"kind": "oauth", "namespace": "issuer/client-b", "value": "same-subject"}
|
|
a = stitch_after_anonymous(client, first)
|
|
client.post("/v1/identity/anonymous?rotate=true")
|
|
b = stitch(client, [second])
|
|
assert a["master_id"] != b["master_id"]
|
|
resolved = client.post(
|
|
"/v1/identity/resolve", headers=AUTH, json={"identifiers": [first, second]}
|
|
).json()
|
|
assert resolved["ambiguous"] is True
|
|
assert resolved["master_id"] is None
|
|
|
|
|
|
def test_email_domain_normalization_preserves_case_sensitive_local_parts(client):
|
|
first = stitch_after_anonymous(client, email("Alice@EXAMPLE.com"))
|
|
resolved = client.post(
|
|
"/v1/identity/resolve",
|
|
headers=AUTH,
|
|
json={"identifiers": [email("Alice@example.com")]},
|
|
).json()
|
|
assert resolved["master_id"] == first["master_id"]
|
|
different = client.post(
|
|
"/v1/identity/resolve",
|
|
headers=AUTH,
|
|
json={"identifiers": [email("alice@example.com")]},
|
|
).json()
|
|
assert different["master_id"] is None
|
|
|
|
|
|
def test_stitch_requires_trusted_backend_and_signed_anonymous_proof(client):
|
|
anon = anonymous(client)
|
|
response = client.post("/v1/identity/stitch", json={"identifiers": [email()]})
|
|
assert response.status_code == 401
|
|
response = client.post(
|
|
"/v1/identity/stitch",
|
|
headers=AUTH,
|
|
json={"identifiers": [email()], "anonymous_token": anon["anonymous_id"]},
|
|
)
|
|
assert response.status_code == 401
|
|
response = client.get("/v1/users/" + anon["master_id"])
|
|
assert response.status_code == 401
|
|
response = client.post(
|
|
"/v1/identity/stitch",
|
|
headers=AUTH,
|
|
json={"identifiers": [{**email(), "verified": True}]},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"identifier",
|
|
[
|
|
email("not-an-email"),
|
|
{"kind": "phone", "value": "4155552671"},
|
|
{"kind": "phone", "value": "+10000000000"},
|
|
{"kind": "oauth", "value": "subject"},
|
|
{"kind": "maid", "value": "device-id"},
|
|
],
|
|
)
|
|
def test_invalid_or_weak_identifiers_cannot_create_edges(client, identifier):
|
|
anon = anonymous(client)
|
|
response = client.post(
|
|
"/v1/identity/stitch", headers=AUTH, json={"identifiers": [identifier]}
|
|
)
|
|
assert response.status_code == 422
|
|
assert (
|
|
client.get("/v1/identity/" + anon["master_id"] + "/graph", headers=AUTH).json()[
|
|
"identifiers"
|
|
]
|
|
== []
|
|
)
|
|
|
|
|
|
def test_browser_origin_and_credentials(client):
|
|
denied = client.post(
|
|
"/v1/events",
|
|
headers={"Origin": "https://untrusted.example"},
|
|
json={"name": "page.view"},
|
|
)
|
|
assert denied.status_code == 403
|
|
preflight = client.options(
|
|
"/v1/events",
|
|
headers={
|
|
"Origin": "http://localhost:5174",
|
|
"Access-Control-Request-Method": "POST",
|
|
},
|
|
)
|
|
assert preflight.status_code == 200
|
|
assert preflight.headers["access-control-allow-credentials"] == "true"
|
|
assert preflight.headers["access-control-allow-origin"] == "http://localhost:5174"
|
|
|
|
|
|
def test_event_replay_is_idempotent_and_cannot_steal_another_subjects_event(client):
|
|
event_id = str(uuid4())
|
|
body = {"event_id": event_id, "name": "page.view"}
|
|
first = client.post("/v1/events", json=body).json()
|
|
replay = client.post("/v1/events", json=body).json()
|
|
assert replay["duplicate"] is True
|
|
assert profile(client, first["master_id"])["event_count"] == 1
|
|
client.post("/v1/identity/anonymous?rotate=true")
|
|
assert client.post("/v1/events", json=body).status_code == 409
|
|
assert (
|
|
client.post(
|
|
"/v1/events",
|
|
json={"name": "page.view", "occurred_at": "2020-01-01T00:00:00"},
|
|
).status_code
|
|
== 422
|
|
)
|
|
|
|
|
|
def test_profile_patch_omits_nulls_and_deletes_explicitly(client):
|
|
anon = anonymous(client)
|
|
client.patch(
|
|
"/v1/users/" + anon["master_id"],
|
|
headers=AUTH,
|
|
json={"attributes": {"keep": 1, "remove": "later"}},
|
|
)
|
|
result = client.patch(
|
|
"/v1/users/" + anon["master_id"],
|
|
headers=AUTH,
|
|
json={"attributes": {"remove": None}},
|
|
).json()
|
|
assert result["attributes"] == {"keep": 1}
|
|
|
|
|
|
def test_persistence_survives_restart(settings):
|
|
with TestClient(create_app(settings)) as first:
|
|
anon = anonymous(first)
|
|
first.post("/v1/events", json={"name": "page.view"})
|
|
stitched = stitch(first, [email()])
|
|
with TestClient(create_app(settings)) as second:
|
|
recorded = second.post(
|
|
"/v1/events",
|
|
json={"name": "page.return", "anonymous_token": anon["anonymous_token"]},
|
|
).json()
|
|
assert recorded["master_id"] == stitched["master_id"]
|
|
assert profile(second, stitched["master_id"])["event_count"] == 2
|
|
service = second.app.state.identity_service
|
|
with service.db.transaction() as conn:
|
|
stored = conn.execute(select(identifiers)).mappings().one()
|
|
assert "alice" not in stored["digest"]
|
|
assert "value" not in stored
|
|
|
|
|
|
def test_tenants_cannot_resolve_or_reuse_each_others_ids(settings):
|
|
other = Settings(
|
|
database_url=settings.database_url,
|
|
tenant_id="other",
|
|
identity_secret=SECRET,
|
|
service_token=SERVICE_TOKEN,
|
|
cookie_secure=False,
|
|
event_registry_required=False,
|
|
)
|
|
with (
|
|
TestClient(create_app(settings)) as first,
|
|
TestClient(create_app(other)) as second,
|
|
):
|
|
a = stitch_after_anonymous(first, email())
|
|
b = stitch_after_anonymous(second, email())
|
|
assert a["master_id"] != b["master_id"]
|
|
assert (
|
|
second.get("/v1/users/" + a["master_id"], headers=AUTH).status_code == 404
|
|
)
|
|
assert (
|
|
second.post(
|
|
"/v1/events",
|
|
json={"name": "forged", "anonymous_token": a["anonymous_token"]},
|
|
).status_code
|
|
== 401
|
|
)
|
|
|
|
|
|
def test_parallel_workers_stitch_same_email_into_one_master(settings):
|
|
databases = [Database(settings.database_url, settings.tenant_id) for _ in range(2)]
|
|
for db in databases:
|
|
db.initialize()
|
|
services = [IdentityService(settings, db) for db in databases]
|
|
tokens = [service.anonymous()["anonymous_token"] for service in services]
|
|
request = StitchRequest(identifiers=[email()])
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(
|
|
pool.map(lambda i: services[i].stitch(tokens[i], request), [0, 1])
|
|
)
|
|
assert results[0]["master_id"] == results[1]["master_id"]
|
|
assert len(services[0].graph(results[0]["master_id"])["identifiers"]) == 1
|
|
finally:
|
|
for db in databases:
|
|
db.close()
|
|
|
|
|
|
def test_merge_limit_failure_rolls_back_all_alias_and_profile_changes(client):
|
|
a = stitch_after_anonymous(client, email())
|
|
client.patch(
|
|
"/v1/users/" + a["master_id"],
|
|
headers=AUTH,
|
|
json={"attributes": {"a" + str(i): i for i in range(150)}},
|
|
)
|
|
client.post("/v1/identity/anonymous?rotate=true")
|
|
b = anonymous(client)
|
|
client.patch(
|
|
"/v1/users/" + b["master_id"],
|
|
headers=AUTH,
|
|
json={"attributes": {"b" + str(i): i for i in range(150)}},
|
|
)
|
|
result = client.post(
|
|
"/v1/identity/stitch", headers=AUTH, json={"identifiers": [email()]}
|
|
)
|
|
assert result.status_code == 409
|
|
assert profile(client, b["master_id"])["master_id"] == b["master_id"]
|
|
assert len(profile(client, a["master_id"])["attributes"]) == 150
|
|
assert len(profile(client, b["master_id"])["attributes"]) == 150
|
|
|
|
|
|
def test_missing_configuration_disables_identity_but_keeps_gateway_available(tmp_path):
|
|
with TestClient(
|
|
create_app(Settings(database_url="sqlite:///" + str(tmp_path / "unused.db")))
|
|
) as client:
|
|
assert client.get("/health").status_code == 200
|
|
assert client.post("/v1/identity/anonymous").status_code == 503
|
|
assert (
|
|
client.post(
|
|
"/v1/identity/stitch", json={"identifiers": [email()]}
|
|
).status_code
|
|
== 503
|
|
)
|
|
# Advertiser auth initializes storage independently of visitor identity.
|
|
assert (tmp_path / "unused.db").exists()
|
|
assert client.get("/v1/advertiser/me").status_code == 401
|
|
|
|
|
|
def test_secure_cookie_configuration():
|
|
with TestClient(
|
|
create_app(Settings(database_url="sqlite://", identity_secret=SECRET)),
|
|
base_url="https://testserver",
|
|
) as client:
|
|
result = client.post("/v1/identity/anonymous")
|
|
assert "Secure" in result.headers["set-cookie"]
|
|
assert anonymous(client)["master_id"] == result.json()["master_id"]
|
|
with pytest.raises(ValueError):
|
|
Settings(cookie_secure=False, cookie_samesite="none")
|
|
with pytest.raises(ValueError):
|
|
Settings(profile_backend="unknown")
|
|
|
|
|
|
def test_parallel_workers_initialize_empty_database_safely(settings):
|
|
databases = [Database(settings.database_url, settings.tenant_id) for _ in range(2)]
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
list(pool.map(lambda db: db.initialize(), databases))
|
|
assert IdentityService(settings, databases[0]).anonymous()["master_id"]
|
|
finally:
|
|
for db in databases:
|
|
db.close()
|