"""
Push to a phone: the encryption, the endpoint a member supplies, and — the
point of the whole feature — that the two switches a person sees are honoured.

Every flow test is two accounts, the one causing a notification and the one
receiving it, because a one-member test proves a one-member property.
"""

import base64
import hashlib
import json

import httpx
import pytest
from cryptography.hazmat.primitives import hashes, hmac, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from membership import add_member
from meshbay_hub import webpush
from meshbay_hub.api import push
from meshbay_hub.db.models import Notification, PushSubscription, User
from sqlalchemy import select


def _b64(s: str) -> bytes:
    s = "".join(s.split())
    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))


def _b64e(b: bytes) -> str:
    return base64.urlsafe_b64encode(b).rstrip(b"=").decode()


def _auth_key(password: str, username: str) -> str:
    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
    return base64.b64encode(
        hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()


async def _user(client, username, password="a-long-enough-passphrase"):
    await client.post("/v1/users/register", json={
        "username": username, "email": f"{username}@example.com",
        "auth_key": _auth_key(password, username)})
    r = await client.post("/v1/users/login", json={
        "username": username, "auth_key": _auth_key(password, username)})
    return r.json()["access_token"]


def _phone():
    """A user agent's keys, as a distributor's connector would generate them."""
    sk = ec.generate_private_key(ec.SECP256R1())
    pk = sk.public_key().public_bytes(
        serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
    auth = b"0123456789abcdef"
    return sk, _b64e(pk), _b64e(auth)


def _hm(key, data):
    h = hmac.HMAC(key, hashes.SHA256())
    h.update(data)
    return h.finalize()


def _decrypt(body: bytes, ua_private, auth_secret: bytes) -> bytes:
    """RFC 8291 from the receiving side, written from the RFC and not from webpush.py."""
    salt, idlen = body[:16], body[20]
    as_public = body[21:21 + idlen]
    ua_public = ua_private.public_key().public_bytes(
        serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
    ecdh = ua_private.exchange(
        ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), as_public))
    ikm = _hm(_hm(auth_secret, ecdh), b"WebPush: info\x00" + ua_public + as_public + b"\x01")
    prk = _hm(salt, ikm)
    cek = _hm(prk, b"Content-Encoding: aes128gcm\x00\x01")[:16]
    nonce = _hm(prk, b"Content-Encoding: nonce\x00\x01")[:12]
    plain = AESGCM(cek).decrypt(nonce, body[21 + idlen:], None)
    assert plain.endswith(b"\x02")
    return plain[:-1]


@pytest.fixture
def sent(monkeypatch):
    """Every push the hub would have sent, instead of sending it."""
    calls = []

    async def fake_send(target, payload, *, ttl, client=None):
        calls.append((target, payload, ttl))
        return webpush.DELIVERED

    monkeypatch.setattr(push, "_send", fake_send)
    push._last_chat.clear()
    return calls


async def _subscribe(client, token, endpoint="https://push.example.net/up/abc"):
    _, p256dh, auth = _phone()
    r = await client.post("/v1/push/subscriptions",
                          json={"endpoint": endpoint, "p256dh": p256dh, "auth": auth},
                          headers={"Authorization": f"Bearer {token}"})
    return r


async def _two_in_a_group(client, db_session, name):
    member = await _user(client, f"{name}_member")
    owner = await _user(client, f"{name}_owner")
    g = await client.post("/v1/groups", json={"name": name},
                          headers={"Authorization": f"Bearer {owner}"})
    gid = g.json()["group_id"]
    await add_member(client, gid, f"{name}_member", {"Authorization": f"Bearer {owner}"})
    uid = (await db_session.execute(
        select(User.id).where(User.username == f"{name}_member"))).scalar_one()
    return member, uid, gid


# ── Encryption ───────────────────────────────────────────────────────────────

def test_encryption_reproduces_the_rfc_8291_example():
    as_private = ec.derive_private_key(
        int.from_bytes(_b64("yfWPiYE-n46HLnH0KqZOF1fJJU3MYrct3AELtAQ-oRw"), "big"),
        ec.SECP256R1())
    out = webpush.encrypt(
        _b64("V2hlbiBJIGdyb3cgdXAsIEkgd2FudCB0byBiZSBhIHdhdGVybWVsb24"),
        _b64("BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcx"
             "aOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4"),
        _b64("BTBZMqHH6r4Tts7J_aSIgg"),
        as_private=as_private, salt=_b64("DGv6ra1nlYgDCS1FRnbzlw"))
    header = _b64("DGv6ra1nlYgDCS1FRnbzlwAAEABBBP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27ml"
                  "mlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A8")
    ciphertext = _b64("8pfeW0KbunFT06SuDKoJH9Ql87S1QUrdirN6GcG7sFz1y1sqLgVi1VhjVkHsUoEs"
                      "bI_0LpXMuGvnzQ")
    assert out == header + ciphertext


def test_a_fresh_encryption_opens_on_the_phone():
    sk, p256dh, auth = _phone()
    body = webpush.encrypt(b'{"kind":"x"}', _b64(p256dh), _b64(auth))
    assert _decrypt(body, sk, _b64(auth)) == b'{"kind":"x"}'


# ── The endpoint is a URL a member chose, and the hub fetches it ─────────────

@pytest.mark.parametrize("endpoint", [
    "http://push.example.net/up/abc",
    "https://127.0.0.1/up",
    "https://10.1.2.3/up",
    "https://[::1]/up",
    "https://169.254.169.254/latest",
    "https://user:pw@push.example.net/up",
    "ftp://push.example.net/up",
])
@pytest.mark.asyncio
async def test_an_endpoint_the_hub_should_not_fetch_is_refused(client, endpoint):
    token = await _user(client, "ssrf_shape")
    r = await _subscribe(client, token, endpoint)
    assert r.status_code == 422, (endpoint, r.text)


@pytest.mark.asyncio
async def test_a_name_resolving_to_a_private_address_is_never_sent_to():
    _, p256dh, auth = _phone()
    seen = []
    mock = httpx.AsyncClient(transport=httpx.MockTransport(
        lambda req: seen.append(req) or httpx.Response(201)))
    result = await webpush.send(webpush.Target("https://localhost/up", p256dh, auth),
                                {"v": 1}, ttl=60, client=mock)
    assert result == webpush.FAILED and seen == []


@pytest.mark.asyncio
async def test_the_request_goes_to_the_address_that_was_checked(monkeypatch):
    """The hostname is the TLS name and the Host header only: a second lookup
    returning something else would never be made."""
    sk, p256dh, auth = _phone()

    async def resolved(host, port):
        assert host == "push.example.net"
        return "93.184.215.14"

    monkeypatch.setattr(webpush, "_resolve_public", resolved)
    seen = []
    mock = httpx.AsyncClient(transport=httpx.MockTransport(
        lambda req: seen.append(req) or httpx.Response(201)))
    result = await webpush.send(
        webpush.Target("https://push.example.net/up/abc?up=1", p256dh, auth),
        {"title": "hello"}, ttl=60, client=mock)
    assert result == webpush.DELIVERED
    (req,) = seen
    assert req.url.host == "93.184.215.14" and req.url.path == "/up/abc"
    assert req.headers["host"] == "push.example.net"
    assert req.extensions["sni_hostname"] == "push.example.net"
    assert req.headers["content-encoding"] == "aes128gcm" and req.headers["ttl"] == "60"
    assert json.loads(_decrypt(req.content, sk, _b64(auth))) == {"title": "hello"}


@pytest.mark.asyncio
async def test_bad_keys_are_refused(client):
    token = await _user(client, "bad_keys_user")
    r = await client.post("/v1/push/subscriptions", json={
        "endpoint": "https://push.example.net/up/abc",
        "p256dh": _b64e(b"\x04" + b"\x01" * 64), "auth": _b64e(b"0" * 16)},
        headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 422


# ── Rows ─────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_registering_again_updates_the_same_row(client):
    token = await _user(client, "registers_again")
    first = (await _subscribe(client, token)).json()["id"]
    second = (await _subscribe(client, token)).json()["id"]
    assert first == second


@pytest.mark.asyncio
async def test_an_account_holds_a_bounded_number_of_subscriptions(client):
    token = await _user(client, "many_phones")
    for i in range(push.MAX_SUBSCRIPTIONS):
        r = await _subscribe(client, token, f"https://push.example.net/up/{i}")
        assert r.status_code == 200, r.text
    r = await _subscribe(client, token, "https://push.example.net/up/one-more")
    assert r.status_code == 429


@pytest.mark.asyncio
async def test_only_the_owner_can_remove_a_subscription(client):
    mine = await _user(client, "sub_owner")
    other = await _user(client, "sub_other")
    sid = (await _subscribe(client, mine)).json()["id"]
    r = await client.delete(f"/v1/push/subscriptions/{sid}",
                            headers={"Authorization": f"Bearer {other}"})
    assert r.status_code == 404
    r = await client.delete(f"/v1/push/subscriptions/{sid}",
                            headers={"Authorization": f"Bearer {mine}"})
    assert r.status_code == 200


# ── What is pushed, and what is not ──────────────────────────────────────────

@pytest.mark.asyncio
async def test_a_notification_reaches_the_phone(client, db_session, sent):
    from meshbay_hub.api.notifications import create_notification

    member, uid, gid = await _two_in_a_group(client, db_session, "pushed")
    assert (await _subscribe(client, member)).status_code == 200
    await create_notification(db_session, uid, "chat_message", "owner posted in pushed",
                              link=f"#/group/{gid}", group_id=gid, aggregate=True)
    await db_session.commit()
    await push.drain()
    (target, payload, ttl) = sent[0]
    assert payload["kind"] == "chat_message" and payload["group_id"] == gid
    assert payload["title"] == "owner posted in pushed" and ttl == webpush.TTL_CHAT


@pytest.mark.asyncio
async def test_a_muted_group_pushes_nothing(client, db_session, sent):
    from meshbay_hub.api.notifications import create_notification

    member, uid, gid = await _two_in_a_group(client, db_session, "hushed")
    await _subscribe(client, member)
    r = await client.post(f"/v1/groups/{gid}/mute", json={"muted": True},
                          headers={"Authorization": f"Bearer {member}"})
    assert r.status_code == 200
    await create_notification(db_session, uid, "chat_message", "owner posted in hushed",
                              group_id=gid, aggregate=True)
    await db_session.commit()
    await push.drain()
    assert sent == []


@pytest.mark.asyncio
async def test_every_notification_turned_off_pushes_nothing_and_stores_nothing(
        client, db_session, sent):
    """The account-wide switch was read by the interface only: rows went on
    being created and hidden. With a phone told about each row, that would have
    been a switch that did nothing."""
    from meshbay_hub.api.notifications import create_notification

    member, uid, gid = await _two_in_a_group(client, db_session, "silenced")
    await _subscribe(client, member)
    r = await client.put("/v1/users/me/preferences/notifications_disabled",
                         json={"value": "true"},
                         headers={"Authorization": f"Bearer {member}"})
    assert r.status_code == 200
    for kind, group in (("chat_message", gid), ("group_invite", None)):
        made = await create_notification(db_session, uid, kind, "something",
                                         group_id=group, aggregate=group is not None)
        assert made is None, kind
    await db_session.commit()
    await push.drain()
    assert sent == []
    rows = (await db_session.execute(
        select(Notification).where(Notification.user_id == uid,
                                   Notification.title == "something"))).scalars().all()
    assert rows == []

    await client.put("/v1/users/me/preferences/notifications_disabled",
                     json={"value": "false"},
                     headers={"Authorization": f"Bearer {member}"})
    await create_notification(db_session, uid, "group_invite", "back on")
    await db_session.commit()
    await push.drain()
    assert [p["title"] for _, p, _ in sent] == ["back on"]


@pytest.mark.asyncio
async def test_a_busy_conversation_reaches_a_phone_once_per_window(client, db_session, sent):
    from meshbay_hub.api.notifications import create_notification

    member, uid, gid = await _two_in_a_group(client, db_session, "chatty")
    await _subscribe(client, member)
    for i in range(5):
        await create_notification(db_session, uid, "chat_message", f"line {i}",
                                  group_id=gid, aggregate=True)
    await create_notification(db_session, uid, "group_invite", "not chat")
    await db_session.commit()
    await push.drain()
    assert [p["title"] for _, p, _ in sent] == ["line 0", "not chat"]


@pytest.mark.asyncio
async def test_a_registration_the_push_server_dropped_falls_back_to_fetching(
        client, db_session, monkeypatch):
    """The row loses its endpoint, not its existence: the phone keeps fetching."""
    from meshbay_hub.api.notifications import create_notification

    async def gone(target, payload, *, ttl, client=None):
        return webpush.GONE

    monkeypatch.setattr(push, "_send", gone)
    push._last_chat.clear()
    member, uid, gid = await _two_in_a_group(client, db_session, "dropped")
    await _subscribe(client, member)
    await create_notification(db_session, uid, "group_invite", "anyone there")
    await db_session.commit()
    await push.drain()
    db_session.expire_all()
    (row,) = (await db_session.execute(
        select(PushSubscription).where(PushSubscription.user_id == uid))).scalars().all()
    assert row.endpoint is None and row.p256dh is None and row.poll_hash is not None


# ── Fetching, for a phone with no distributor ───────────────────────────────

async def _poll(client, reg, since=None):
    return await client.post("/v1/push/poll", json={
        "id": reg["id"], "secret": reg["poll_secret"], "since": since or reg["now"]})


@pytest.fixture
def unthrottled(monkeypatch):
    monkeypatch.setattr(push, "POLL_MIN_INTERVAL", 0.0)
    push._last_poll.clear()


@pytest.mark.asyncio
async def test_a_phone_without_a_distributor_fetches_what_is_new(
        client, db_session, sent, unthrottled):
    from meshbay_hub.api.notifications import create_notification

    member, uid, gid = await _two_in_a_group(client, db_session, "fetched")
    r = await client.post("/v1/push/subscriptions", json={},
                          headers={"Authorization": f"Bearer {member}"})
    assert r.status_code == 200, r.text
    reg = r.json()
    # The invitation that made them a member predates the registration: not news.
    assert (await _poll(client, reg)).json()["notifications"] == []

    await create_notification(db_session, uid, "chat_message", "owner posted in fetched",
                              link=f"#/group/{gid}", group_id=gid, aggregate=True)
    await db_session.commit()
    await push.drain()
    assert sent == [], "a row with no endpoint is never pushed to"
    (got,) = (await _poll(client, reg)).json()["notifications"]
    assert got["kind"] == "chat_message" and got["group_id"] == gid
    assert got["title"] == "owner posted in fetched"
    assert (await _poll(client, reg, got["created_at"])).json()["notifications"] == []


@pytest.mark.asyncio
async def test_fetching_honours_both_switches(client, db_session, unthrottled):
    from meshbay_hub.api.notifications import create_notification

    member, uid, gid = await _two_in_a_group(client, db_session, "fetchmute")
    reg = (await client.post("/v1/push/subscriptions", json={},
                             headers={"Authorization": f"Bearer {member}"})).json()
    await client.post(f"/v1/groups/{gid}/mute", json={"muted": True},
                      headers={"Authorization": f"Bearer {member}"})
    await create_notification(db_session, uid, "chat_message", "muted line",
                              group_id=gid, aggregate=True)
    await client.put("/v1/users/me/preferences/notifications_disabled",
                     json={"value": "true"}, headers={"Authorization": f"Bearer {member}"})
    await create_notification(db_session, uid, "group_invite", "all off")
    await db_session.commit()
    assert (await _poll(client, reg)).json()["notifications"] == []


@pytest.mark.asyncio
async def test_the_poll_secret_is_the_only_way_in(client, unthrottled):
    mine = await _user(client, "poll_owner")
    reg = (await client.post("/v1/push/subscriptions", json={},
                             headers={"Authorization": f"Bearer {mine}"})).json()
    assert (await _poll(client, {**reg, "poll_secret": "x" * 43})).status_code == 404
    assert (await _poll(client, {**reg, "id": "0" * 36})).status_code == 404
    again = (await client.post("/v1/push/subscriptions", json={"id": reg["id"]},
                               headers={"Authorization": f"Bearer {mine}"})).json()
    assert again["id"] == reg["id"]
    assert (await _poll(client, reg)).status_code == 404, "a new secret retires the old"
    assert (await _poll(client, again)).status_code == 200
    await client.delete(f"/v1/push/subscriptions/{reg['id']}",
                        headers={"Authorization": f"Bearer {mine}"})
    assert (await _poll(client, again)).status_code == 404, "signed out: the row is gone"


@pytest.mark.asyncio
async def test_another_account_cannot_take_over_a_row(client):
    mine = await _user(client, "row_owner")
    other = await _user(client, "row_taker")
    reg = (await client.post("/v1/push/subscriptions", json={},
                             headers={"Authorization": f"Bearer {mine}"})).json()
    theirs = (await client.post("/v1/push/subscriptions", json={"id": reg["id"]},
                                headers={"Authorization": f"Bearer {other}"})).json()
    assert theirs["id"] != reg["id"]


@pytest.mark.asyncio
async def test_a_phone_cannot_poll_faster_than_the_floor(client):
    push._last_poll.clear()
    mine = await _user(client, "eager_poller")
    reg = (await client.post("/v1/push/subscriptions", json={},
                             headers={"Authorization": f"Bearer {mine}"})).json()
    assert (await _poll(client, reg)).status_code == 200
    r = await _poll(client, reg)
    assert r.status_code == 429 and int(r.headers["retry-after"]) > 0


@pytest.mark.asyncio
async def test_a_phone_gaining_a_distributor_keeps_its_row(client):
    mine = await _user(client, "upgrading_phone")
    reg = (await client.post("/v1/push/subscriptions", json={},
                             headers={"Authorization": f"Bearer {mine}"})).json()
    _, p256dh, auth = _phone()
    r = await client.post("/v1/push/subscriptions", json={
        "id": reg["id"], "endpoint": "https://push.example.net/up/new",
        "p256dh": p256dh, "auth": auth}, headers={"Authorization": f"Bearer {mine}"})
    assert r.json()["id"] == reg["id"]


@pytest.mark.asyncio
async def test_an_endpoint_without_its_keys_is_refused(client):
    mine = await _user(client, "keyless_phone")
    r = await client.post("/v1/push/subscriptions",
                          json={"endpoint": "https://push.example.net/up/k"},
                          headers={"Authorization": f"Bearer {mine}"})
    assert r.status_code == 422