import json
import time
from urllib.parse import parse_qs
from urllib.parse import urlparse

import pytest

from oic import rndstr
from oic.extension.client import Client
from oic.extension.message import TokenIntrospectionRequest
from oic.extension.message import TokenIntrospectionResponse
from oic.extension.message import TokenRevocationRequest
from oic.extension.provider import Provider
from oic.extension.token import JWTToken
from oic.oauth2.message import AccessTokenRequest
from oic.oauth2.message import AccessTokenResponse
from oic.oauth2.message import AuthorizationRequest
from oic.oauth2.message import AuthorizationResponse
from oic.oauth2.message import ROPCAccessTokenRequest
from oic.oauth2.message import TokenErrorResponse
from oic.utils.authn.authn_context import AuthnBroker
from oic.utils.authn.client import verify_client
from oic.utils.authn.user import UserAuthnMethod
from oic.utils.authz import Implicit
from oic.utils.keyio import KeyBundle
from oic.utils.keyio import KeyJar
from oic.utils.sdb import DefaultToken
from oic.utils.sdb import SessionDB
from oic.utils.sdb import lv_pack
from oic.utils.sdb import lv_unpack

CLIENT_CONFIG = {"client_id": "client1", "config": {"issuer": "https://example.com/as"}}

CONSUMER_CONFIG = {
    "authz_page": "/authz",
    "flow_type": "code",
    "scope": [],
    "response_type": "code",
}

ISSUER = "https://connect-op.heroku.com"
SERVER_INFO = {
    "version": "3.0",
    "issuer": ISSUER,
    "authorization_endpoint": "http://localhost:8088/authorization",
    "token_endpoint": "http://localhost:8088/token",
    "flows_supported": ["code", "token", "code token"],
}

CDB = {
    "a1b2c3": {
        "password": "hemligt",
        "client_secret": "drickyoughurt",
        "redirect_uris": [("http://localhost:8087/authz", None)],
        "token_endpoint_auth_method": "client_secret_post",
        "response_types": ["code", "token"],
    },
    "client1": {
        "client_secret": "hemlighet",
        "redirect_uris": [("http://localhost:8087/authz", None)],
        "token_endpoint_auth_method": "client_secret_post",
        "response_types": ["code", "token"],
    },
}

JWKS = {
    "keys": [
        {
            "d": "vT9bnSZ63uIdaVsmZjrbmcvrDZG-_qzVQ1KmrSSC398sLJiyaQKRPkmBRvV"
            "-MGxW1MVPeCkhnSULCRgtqHq"
            "-zQxMeCviSScHTKOuDYJfwMB5qdOE3FkuqPMsEVf6EXYaSd90"
            "-O6GOA88LBCPNR4iKxsrQ6LNkawwiJoPw7muK3TbQk9HzuznF8WDkt72CQFxd4eT"
            "6wJ97xpaIgxZce0oRmFcLYkQ4A0pgVhF42zxJjJDIBj_ZrSl5_qZIgiE76PV4hjH"
            "t9Nv4ZveabObnNbyz9YOiWHiOLdYZGmixHuauM98NK8udMxI6IuOkRypFhJzaQZF"
            "wMroa7ZNZF-mm78VYQ",
            "dp": "wLqivLfMc0FBhGFFRTb6WWzDpVZukcgOEQGb8wW3knmNEpgch699WQ4ZY_ws1xSbv"
            "QZtbx7MaIBXpn3qT1LYZosoP5oHVTAvdg6G8I7zgWyqj-nG4evciuoeAa1Ff52h4-"
            "J1moZ6FF2GelLdjXHoCbjIBjz_VljelSqOk5Sh5HU",
            "dq": "KXIUYNfDxwxv3A_w1t9Ohm92gOs-UJdI3_IVpe4FauCDrJ4mqgsnTisA15KY"
            "-9fCEvKfqG571WK6EKpBcxaRrqSU0ekpBvgJx8o3MGlqXWj-Lw0co8N9_"
            "-fo1rYx_8g-wCRrm5zeA5pYJdwdhOBnmKOqw_GsXJEcYeUod1xkcfU",
            "e": "AQAB",
            "ext": "true",
            "key_ops": "sign",
            "kty": "RSA",
            "n": "wl0DPln-EFLqr_Ftn6A87wEQAUVbpZsUTN2OCEsJV0nhlvmX3GUzyZx5UXdlM3Dz68PfUWCgfx67Il6sURqWVCnjnU-"
            "_gr3GeDyzedj-lZejnBx-lEy_3j6B98SbcDfkJF6saXnPd7_kgilJT1_g-EVI9ifFB1cxZXHCd2WBeRABSCprAlCglF-YmnU"
            "eeDs5K32z2ckVjadF9BG27CO5UfNq0K8jI9Yj_coOhM9dRNrQ9UVZNdQVG-bAIDhB2y2o3ASGwqchHouIxv5YZNGS0SMJL5t"
            "0edh483q1tSWPqBw-ZeryLztOedBBzSuJk7QDmL1B6B7KKUIrlUYJmVsYzw",
            "p": "6MEg5Di_IFiPGKvMFRjyx2t7YAOQ4KfdIkU_Khny1t1eCG5O07omPe_jLU8I5fPaD5F5HhWExLNureHD4K6LB18JPE3"
            "VE8chQROiRSNPZo1-faUvHu-Dy0pr7I-TS8pl_P3vop1KelIbGwXhzPIRKQMqCEKi3tLJt4R_MQ18Dx0",
            "q": "1cZVPpUbf4p5n4cMv_kERCPh3cieMs4aVojgh3feAiJiLwWWL9Pc43oJUekK44aWMnbs68Y4kqXtc52PMtBDzVp0Gjt"
            "0lCY3M7MYRVI4JhtknqvQynMKQ2nKs3VldvVfY2SxyUmnRyEolQUGRA7rRMUyPb4AXhSR7oroRrJD59s",
            "qi": "50PhyaqbLSczhipWiYy149sLsGlx9cX0tnGMswy1JLam7nBvH4"
            "-MWB2oGwD2hmG-YN66q-xXBS9CVDLZZrj1sonRTQPtWE"
            "-zuZqds6_NVlk2Ge4_IAA3TZ9tvIfM5FZVTOQsExu3_LX8FGCspWC1R"
            "-zDqT45Y9bpaCwxekluO7Q",
            "kid": "sign1",
        },
        {
            "k": b"YTEyZjBlMDgxMGI4YWU4Y2JjZDFiYTFlZTBjYzljNDU3YWM0ZWNiNzhmNmFlYTNkNTY0NzMzYjE",
            "kty": "oct",
            "use": "sig",
        },
    ]
}


def _eq(l1, l2):
    return set(l1) == set(l2)


def test_lv_pack_unpack():
    lst = ["foo", "kaka", "banan", "jordgubb"]
    s = lv_pack(*lst)
    r = lv_unpack(s)
    assert r == lst


class DummyAuthn(UserAuthnMethod):
    def __init__(self, srv, user, password=None):
        UserAuthnMethod.__init__(self, srv)
        self.user = user
        self.password = password

    def authenticated_as(self, cookie=None, **kwargs):
        if self.password is not None and kwargs.get("password") != self.password:
            return None, 0
        return {"uid": self.user}, time.time()


AUTHN_BROKER = AuthnBroker()
AUTHN_BROKER.add("UNDEFINED", DummyAuthn(None, "username"))
AUTHN_BROKER2 = AuthnBroker()
AUTHN_BROKER2.add("UNDEFINED", DummyAuthn(None, "username", "password"))
# dealing with authorization
AUTHZ = Implicit()


class TestProvider(object):
    @pytest.fixture(autouse=True)
    def create_provider(self):
        kb = KeyBundle(JWKS["keys"])
        kj = KeyJar()
        kj.issuer_keys[""] = [kb]

        _sdb = SessionDB(
            "https://example.com/",
            db={},
            code_factory=DefaultToken(
                "supersecret", "verybadpassword", typ="A", lifetime=600
            ),
            token_factory=JWTToken(
                "T",
                keyjar=kj,
                lt_pattern={"code": 3600, "token": 900},
                iss="https://example.com/as",
                sign_alg="RS256",
            ),
            refresh_token_factory=JWTToken(
                "R",
                keyjar=kj,
                lt_pattern={"": 24 * 3600},
                iss="https://example.com/as",
                token_storage={},
            ),
        )
        #  name, sdb, cdb, authn_broker, authz, client_authn,
        self.provider = Provider(
            "as",
            _sdb,
            CDB,
            AUTHN_BROKER,
            AUTHZ,
            verify_client,
            baseurl="https://example.com/as",
        )

    def test_authorization_endpoint_faulty_redirect_uri(self):
        bib = {
            "state": "id-6da9ca0cc23959f5f33e8becd9b08cae",
            # faulty redirect uri
            "redirect_uri": "http://localhost:8087/cb",
            "response_type": ["code"],
            "client_id": "a1b2c3",
        }

        arq = AuthorizationRequest(**bib)
        resp = self.provider.authorization_endpoint(request=arq.to_urlencoded())
        assert resp.status_code == 400
        msg = json.loads(resp.message)
        assert msg["error"] == "invalid_request"

    def test_authenticated(self):
        client = Client(**CLIENT_CONFIG)
        client.authorization_endpoint = "https://example.com/as"

        sid = rndstr(8)
        args = {
            "redirect_uri": "http://localhost:8087/authz",
            "state": sid,
            "response_type": "code",
        }

        url, body, ht_args, csi = client.request_info(
            AuthorizationRequest, "GET", request_args=args
        )

        resp = self.provider.authorization_endpoint(urlparse(url).query)
        assert resp.status_code == 303
        resp = urlparse(resp.message).query
        aresp = client.parse_authz_response(resp)

        assert isinstance(aresp, AuthorizationResponse)
        assert _eq(aresp.keys(), ["state", "code", "client_id", "iss"])
        assert _eq(
            client.grant[sid].keys(),
            ["tokens", "code", "exp_in", "seed", "id_token", "grant_expiration_time"],
        )

    def test_authenticated_token(self):
        client = Client(**CLIENT_CONFIG)
        client.authorization_endpoint = "https://example.com/as"

        sid = rndstr(8)
        args = {
            "redirect_uri": "http://localhost:8087/authz",
            "state": sid,
            "response_type": "token",
        }

        url, body, ht_args, csi = client.request_info(
            AuthorizationRequest, "GET", request_args=args
        )

        QUERY_STRING = url.split("?")[1]
        resp = self.provider.authorization_endpoint(QUERY_STRING)
        auth_resp = parse_qs(urlparse(resp.message).fragment)

        assert "access_token" in auth_resp
        assert auth_resp["token_type"][0] == "Bearer"

    def test_token_endpoint(self):
        authreq = AuthorizationRequest(
            state="state", redirect_uri="http://example.com/authz", client_id="client1"
        )

        _sdb = self.provider.sdb
        sid = _sdb.access_token.key(user="sub", areq=authreq)
        access_grant = _sdb.token_factory["code"](sid=sid)
        _sdb[sid] = {
            "oauth_state": "authz",
            "sub": "sub",
            "authzreq": authreq.to_json(),
            "client_id": "client1",
            "code": access_grant,
            "code_used": False,
            "redirect_uri": "http://example.com/authz",
            "response_type": ["code"],
        }

        # Construct Access token request
        areq = AccessTokenRequest(
            code=access_grant,
            redirect_uri="http://example.com/authz",
            client_id="client1",
            client_secret="hemlighet",
            grant_type="authorization_code",
        )

        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
        atr = AccessTokenResponse().deserialize(resp.message, "json")
        assert _eq(atr.keys(), ["access_token", "token_type"])

    def test_token_endpoint_no_cache(self):
        authreq = AuthorizationRequest(
            state="state", redirect_uri="http://example.com/authz", client_id="client1"
        )

        _sdb = self.provider.sdb
        sid = _sdb.access_token.key(user="sub", areq=authreq)
        access_grant = _sdb.token_factory["code"](sid=sid)
        _sdb[sid] = {
            "oauth_state": "authz",
            "sub": "sub",
            "authzreq": authreq.to_json(),
            "client_id": "client1",
            "code": access_grant,
            "code_used": False,
            "redirect_uri": "http://example.com/authz",
            "response_type": ["code"],
        }

        # Construct Access token request
        areq = AccessTokenRequest(
            code=access_grant,
            redirect_uri="http://example.com/authz",
            client_id="client1",
            client_secret="hemlighet",
            grant_type="authorization_code",
        )

        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
        assert resp.headers == [
            ("Pragma", "no-cache"),
            ("Cache-Control", "no-store"),
            ("Content-type", "application/json"),
        ]

    def test_token_endpoint_unauth(self):
        authreq = AuthorizationRequest(
            state="state",
            redirect_uri="http://example.com/authz",
            client_id="client1",
            response_type="code",
        )

        _sdb = self.provider.sdb
        sid = _sdb.access_token.key(user="sub", areq=authreq)
        access_grant = _sdb.token_factory["code"](sid=sid)
        _sdb[sid] = {
            "oauth_state": "authz",
            "sub": "sub",
            "authzreq": authreq.to_json(),
            "client_id": "client1",
            "code": access_grant,
            "code_used": False,
            "redirect_uri": "http://example.com/authz",
            "response_type": ["code"],
        }

        # Construct Access token request
        areq = AccessTokenRequest(
            code=access_grant,
            redirect_uri="http://example.com/authz",
            client_id="client2",
            client_secret="hemlighet",
            grant_type="authorization_code",
        )

        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
        atr = TokenErrorResponse().deserialize(resp.message, "json")
        assert _eq(atr.keys(), ["error_description", "error"])

    def test_token_introspection(self):
        authreq = AuthorizationRequest(
            state="state", redirect_uri="http://example.com/authz", client_id="client1"
        )

        _sdb = self.provider.sdb
        sid = _sdb.access_token.key(user="sub", areq=authreq)
        access_grant = _sdb.token_factory["code"](sid=sid)
        _sdb[sid] = {
            "oauth_state": "authz",
            "sub": "sub",
            "authzreq": authreq.to_json(),
            "client_id": "client1",
            "code": access_grant,
            "code_used": False,
            "redirect_uri": "http://example.com/authz",
            "response_type": ["code"],
        }

        # Construct Access token request
        areq = AccessTokenRequest(
            code=access_grant,
            redirect_uri="http://example.com/authz",
            client_id="client1",
            client_secret="hemlighet",
            grant_type="authorization_code",
        )

        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
        atr = AccessTokenResponse().deserialize(resp.message, "json")
        req = TokenIntrospectionRequest(
            token=atr["access_token"],
            client_id="client1",
            client_secret="hemlighet",
            token_type_hint="access_token",
        )
        resp = self.provider.introspection_endpoint(request=req.to_urlencoded())
        assert resp
        ti_resp = TokenIntrospectionResponse().deserialize(resp.message, "json")
        assert ti_resp["active"] is True

    def test_token_introspection_bad_access_token(self):
        req = TokenIntrospectionRequest(
            token="access_token",
            client_id="client1",
            client_secret="hemlighet",
            token_type_hint="access_token",
        )
        resp = self.provider.introspection_endpoint(request=req.to_urlencoded())
        assert resp
        ti_resp = TokenIntrospectionResponse().deserialize(resp.message, "json")
        assert ti_resp["active"] is False

    def test_token_introspection_bad_token_no_hint(self):
        req = TokenIntrospectionRequest(
            token="access_token", client_id="client1", client_secret="hemlighet"
        )
        resp = self.provider.introspection_endpoint(request=req.to_urlencoded())
        assert resp
        ti_resp = TokenIntrospectionResponse().deserialize(resp.message, "json")
        assert ti_resp["active"] is False

    def test_token_introspection_missing(self):
        authreq = AuthorizationRequest(
            state="state", redirect_uri="http://example.com/authz", client_id="client2"
        )

        _sdb = self.provider.sdb
        self.provider.cdb["client2"] = {
            "client_secret": "hemlighet",
            "redirect_uris": [("http://localhost:8087/authz", None)],
            "token_endpoint_auth_method": "client_secret_post",
            "response_types": ["code", "token"],
        }
        sid = _sdb.access_token.key(user="sub", areq=authreq)
        access_grant = _sdb.token_factory["code"](sid=sid)
        _sdb[sid] = {
            "oauth_state": "authz",
            "sub": "sub",
            "authzreq": authreq.to_json(),
            "client_id": "client2",
            "code": access_grant,
            "code_used": False,
            "redirect_uri": "http://example.com/authz",
            "response_type": ["code"],
        }

        # Construct Access token request
        areq = AccessTokenRequest(
            code=access_grant,
            redirect_uri="http://example.com/authz",
            client_id="client2",
            client_secret="hemlighet",
            grant_type="authorization_code",
        )

        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
        atr = AccessTokenResponse().deserialize(resp.message, "json")
        # Delete the client
        del self.provider.cdb["client2"]
        req = TokenIntrospectionRequest(
            token=atr["access_token"],
            client_id="client2",
            client_secret="hemlighet",
            token_type_hint="access_token",
        )
        resp = self.provider.introspection_endpoint(request=req.to_urlencoded())
        assert resp
        ti_resp = TokenIntrospectionResponse().deserialize(resp.message, "json")
        assert ti_resp["error"] == "unauthorized_client"

    def test_token_revocation_and_introspection(self):
        authreq = AuthorizationRequest(
            state="state", redirect_uri="http://example.com/authz", client_id="client1"
        )

        _sdb = self.provider.sdb
        sid = _sdb.access_token.key(user="sub", areq=authreq)
        access_grant = _sdb.token_factory["code"](sid=sid)
        _sdb[sid] = {
            "oauth_state": "authz",
            "sub": "sub",
            "authzreq": authreq.to_json(),
            "client_id": "client1",
            "code": access_grant,
            "code_used": False,
            "redirect_uri": "http://example.com/authz",
            "response_type": ["code"],
        }

        # Construct Access token request
        areq = AccessTokenRequest(
            code=access_grant,
            redirect_uri="http://example.com/authz",
            client_id="client1",
            client_secret="hemlighet",
            grant_type="authorization_code",
        )

        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
        atr = AccessTokenResponse().deserialize(resp.message, "json")

        req = TokenRevocationRequest(
            token=atr["access_token"],
            client_id="client1",
            client_secret="hemlighet",
            token_type_hint="access_token",
        )
        resp = self.provider.revocation_endpoint(request=req.to_urlencoded())
        assert resp.status_code == 200

        req2 = TokenIntrospectionRequest(
            token=atr["access_token"],
            client_id="client1",
            client_secret="hemlighet",
            token_type_hint="access_token",
        )
        resp = self.provider.introspection_endpoint(request=req2.to_urlencoded())
        assert resp
        ti_resp = TokenIntrospectionResponse().deserialize(resp.message, "json")
        assert ti_resp["active"] is False

    def test_password_grant_type_ok(self):
        # Set a not so dummy Authn method and token policy
        self.provider.authn_broker = AUTHN_BROKER2
        self.provider.set_token_policy("client1", {"grant_type": ["password"]})
        areq = ROPCAccessTokenRequest(
            grant_type="password", username="username", password="password"
        )
        areq[
            "client_id"
        ] = "client1"  # Token endpoint would fill that in based on client_authn
        resp = self.provider.password_grant_type(areq)

        atr = AccessTokenResponse().deserialize(resp.message, "json")
        assert _eq(atr.keys(), ["access_token", "token_type", "refresh_token"])

    def test_password_grant_type_no_authn(self):
        # Set a blank AuthnBroker
        self.provider.authn_broker = AuthnBroker()
        self.provider.set_token_policy("client1", {"grant_type": ["password"]})
        areq = ROPCAccessTokenRequest(
            grant_type="password", username="username", password="password"
        )
        areq[
            "client_id"
        ] = "client1"  # Token endpoint would fill that in based on client_authn
        resp = self.provider.password_grant_type(areq)

        atr = TokenErrorResponse().deserialize(resp.message, "json")
        assert atr["error"] == "invalid_grant"

    def test_password_grant_type_bad(self):
        # Set a not so dummy Authn method and token policy
        self.provider.authn_broker = AUTHN_BROKER2
        self.provider.set_token_policy("client1", {"grant_type": ["password"]})
        areq = ROPCAccessTokenRequest(
            grant_type="password", username="username", password="bad_password"
        )
        areq[
            "client_id"
        ] = "client1"  # Token endpoint would fill that in based on client_authn
        resp = self.provider.password_grant_type(areq)

        atr = TokenErrorResponse().deserialize(resp.message, "json")
        assert atr["error"] == "invalid_grant"
