"""LDAP and SSO sign-in: the directory is reached over checked TLS with the
login name escaped, an SSO token must belong to the sign-in it ends, and a
provider's account never signs in as a local one."""
import base64
import json
import sys
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest

from steward.auth import ldap_auth
from steward.auth.oidc import id_token_refusal
from tests.test_auth import make_user, with_csrf


@pytest.mark.parametrize("raw,escaped", [
    ("bob", "bob"),
    ("*", "\\2a"),
    ("*)(uid=*", "\\2a\\29\\28uid=\\2a"),
    ("a\\b", "a\\5cb"),
    ("nul\x00", "nul\\00"),
])
def test_filter_value_escapes_filter_syntax(raw, escaped):
    assert ldap_auth.filter_value(raw) == escaped


@pytest.mark.parametrize("cfg,want", [
    ({"port": 636}, (True, False)),
    ({"port": 636, "tls": True}, (True, False)),
    ({"port": 389, "tls": True}, (False, True)),
    ({"port": 389}, (False, False)),
])
def test_transport(cfg, want):
    assert ldap_auth.transport(cfg) == want


class _FakeLdap:
    """Just enough of ldap3 to record what Steward asks the directory."""

    ALL = "ALL"
    AUTO_BIND_TLS_BEFORE_BIND = "tls-before-bind"
    AUTO_BIND_NO_TLS = "no-tls"

    def __init__(self, password="pw"):
        self.password = password
        self.servers, self.binds, self.filters, self.tls = [], [], [], []
        fake = self

        class Connection:
            def __init__(self, server, user, password, auto_bind):
                fake.binds.append((user, auto_bind))
                if user.startswith("uid=") and password != fake.password:
                    raise RuntimeError("invalid credentials")
                self.entries = []

            def search(self, search_base, search_filter, attributes):
                fake.filters.append(search_filter)
                self.entries = [SimpleNamespace(entry_dn="uid=bob,ou=people,dc=x", uid=["bob"],
                                                mail=["bob@x"], memberOf=["cn=ops,dc=x"])]

            def unbind(self):
                pass

        self.Connection = Connection

    def Server(self, host, port, use_ssl, tls, get_info):
        self.servers.append({"port": port, "use_ssl": use_ssl, "tls": tls})
        return object()

    def Tls(self, validate, ca_certs_data):
        self.tls.append((validate, ca_certs_data))
        return "tls"


_CFG = {"host": "ldap", "port": 389, "tls": True, "bind_dn": "cn=svc", "base_dn": "dc=x",
        "user_filter": "(&(objectClass=person)(uid={username}))",
        "operator_group_dn": "cn=ops,dc=x", "ca_cert": "PEM"}


def test_ldap_starts_tls_checks_the_certificate_and_escapes_the_name(monkeypatch):
    import ssl
    fake = _FakeLdap()
    monkeypatch.setitem(sys.modules, "ldap3", fake)
    info = ldap_auth._ldap_authenticate_sync(_CFG, "bob*", "pw")
    assert fake.filters == ["(&(objectClass=person)(uid=bob\\2a))"]
    assert fake.servers == [{"port": 389, "use_ssl": False, "tls": "tls"}]
    assert fake.tls == [(ssl.CERT_REQUIRED, "PEM")]
    assert {b[1] for b in fake.binds} == {"tls-before-bind"}
    assert info == {"subject": "uid=bob,ou=people,dc=x", "username": "bob",
                    "email": "bob@x", "role": "operator"}


def test_ldap_wrong_password_is_refused(monkeypatch):
    monkeypatch.setitem(sys.modules, "ldap3", _FakeLdap(password="right"))
    assert ldap_auth._ldap_authenticate_sync(_CFG, "bob", "wrong") is None


@pytest.mark.asyncio
async def test_ldap_empty_password_never_reaches_the_directory(monkeypatch):
    fake = _FakeLdap(password="")
    monkeypatch.setitem(sys.modules, "ldap3", fake)
    assert await ldap_auth.ldap_authenticate(_CFG, "bob", "") is None
    assert fake.binds == []


def _jwt(claims: dict) -> str:
    body = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
    return f"e30.{body}.sig"


@pytest.mark.parametrize("token,nonce,sub,ok", [
    (_jwt({"nonce": "n1", "sub": "u1"}), "n1", "u1", True),
    (_jwt({"nonce": "n2", "sub": "u1"}), "n1", "u1", False),   # another sign-in's token
    (_jwt({"nonce": "n1", "sub": "u1"}), None, "u1", False),   # no sign-in was started
    (_jwt({"nonce": "n1", "sub": "u2"}), "n1", "u1", False),   # token names someone else
    ("", "n1", "u1", False),
    (_jwt({"nonce": "n1", "sub": "u1"}), "n1", "", False),
])
def test_id_token_must_belong_to_this_sign_in(token, nonce, sub, ok):
    assert (id_token_refusal(token, nonce, sub) is None) is ok


@pytest.mark.asyncio
async def test_local_password_does_not_sign_in_a_directory_account(client):
    user = make_user()
    user.auth_source = "ldap"
    headers = await with_csrf(client)
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
         patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=user)):
        response = await client.post("/login", form={"username": "admin", "password": "password"},
                                     headers=headers)
    assert response.status_code == 400


@pytest.mark.asyncio
async def test_ldap_sign_in_shows_why_it_was_refused(app, client):
    app.config["LDAP"] = {"enabled": True}
    headers = await with_csrf(client)
    info = {"subject": "uid=admin,dc=x", "username": "admin", "email": "", "role": "viewer"}
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
         patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=make_user())), \
         patch("steward.auth.ldap_auth.ldap_authenticate", new=AsyncMock(return_value=info)), \
         patch("steward.auth.routes._provision_external_user",
               new=AsyncMock(return_value=(None, "Steward already has an account named admin"))):
        response = await client.post("/login", form={"username": "admin", "password": "dir-pw"},
                                     headers=headers)
    assert response.status_code == 403
    assert "already has an account named admin" in await response.get_data(as_text=True)
    async with client.session_transaction() as sess:
        assert "user_id" not in sess