# authentic2 - versatile identity manager
# Copyright (C) 2010-2021 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import base64
import datetime
import functools
import json
import urllib.parse

import pytest
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.core.files import File
from django.http import QueryDict
from django.test.utils import override_settings
from django.urls import reverse
from django.utils.encoding import force_text
from django.utils.timezone import now
from jwcrypto.jwk import JWK, JWKSet
from jwcrypto.jwt import JWT

from authentic2.a2_rbac.utils import get_default_ou
from authentic2.models import Attribute, AuthorizedRole
from authentic2.utils.misc import good_next_url, make_url
from authentic2_auth_oidc.utils import parse_timestamp
from authentic2_idp_oidc.models import OIDCAccessToken, OIDCAuthorization, OIDCClaim, OIDCClient, OIDCCode
from authentic2_idp_oidc.utils import base64url, get_first_ec_sig_key, get_first_rsa_sig_key, make_sub
from django_rbac.utils import get_ou_model, get_role_model

from .. import utils
from .conftest import bearer_authentication_headers, client_authentication_headers

User = get_user_model()

pytestmark = pytest.mark.django_db


def test_get_jwkset(oidc_settings):
    from authentic2_idp_oidc.utils import get_jwkset

    get_jwkset()


OIDC_CLIENT_PARAMS = [
    {
        'authorization_flow': OIDCClient.FLOW_IMPLICIT,
    },
    {
        'post_logout_redirect_uris': 'https://example.com/',
    },
    {
        'identifier_policy': OIDCClient.POLICY_UUID,
        'post_logout_redirect_uris': 'https://example.com/',
    },
    {
        'identifier_policy': OIDCClient.POLICY_EMAIL,
    },
    {
        'idtoken_algo': OIDCClient.ALGO_HMAC,
    },
    {
        'idtoken_algo': OIDCClient.ALGO_EC,
    },
    {
        'authorization_mode': OIDCClient.AUTHORIZATION_MODE_NONE,
    },
    {
        'idtoken_duration': datetime.timedelta(hours=1),
    },
    {
        'authorization_flow': OIDCClient.FLOW_IMPLICIT,
        'idtoken_duration': datetime.timedelta(hours=1),
        'post_logout_redirect_uris': 'https://example.com/',
    },
    {
        'frontchannel_logout_uri': 'https://example.com/southpark/logout/',
    },
    {
        'frontchannel_logout_uri': 'https://example.com/southpark/logout/',
        'frontchannel_timeout': 3000,
    },
    {
        'identifier_policy': OIDCClient.POLICY_PAIRWISE_REVERSIBLE,
    },
]


@pytest.mark.parametrize('other_attributes', OIDC_CLIENT_PARAMS)
def test_admin(other_attributes, app, superuser, oidc_settings):
    Attribute.objects.create(
        name='cityscape_image',
        label='cityscape',
        kind='profile_image',
        asked_on_registration=True,
        required=False,
        user_visible=True,
        user_editable=True,
    )

    url = reverse('admin:authentic2_idp_oidc_oidcclient_add')
    assert OIDCClient.objects.count() == 0
    response = utils.login(app, superuser, path=url)
    response.form.set('name', 'oidcclient')
    response.form.set('slug', 'oidcclient')
    response.form.set('ou', get_default_ou().pk)
    response.form.set('unauthorized_url', 'https://example.com/southpark/')
    response.form.set('redirect_uris', 'https://example.com/callbac%C3%A9')
    for key, value in other_attributes.items():
        response.form.set(key, value)
    response = response.form.submit().follow()
    assert OIDCClient.objects.count() == 1


@pytest.mark.parametrize('oidc_client', OIDC_CLIENT_PARAMS, indirect=True)
@pytest.mark.parametrize('do_not_ask_again', [(True,), (False,)])
@pytest.mark.parametrize('login_first', [(True,), (False,)])
def test_authorization_code_sso(
    login_first, do_not_ask_again, oidc_client, oidc_settings, simple_user, app, caplog
):
    redirect_uri = oidc_client.redirect_uris.split()[0]
    params = {
        'client_id': oidc_client.client_id,
        'scope': 'openid profile email',
        'redirect_uri': redirect_uri,
        'state': 'xxx',
        'nonce': 'yyy',
        'login_hint': 'backoffice john@example.com',
    }

    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        params['response_type'] = 'code'
    elif oidc_client.authorization_flow == oidc_client.FLOW_IMPLICIT:
        params['response_type'] = 'token id_token'
    authorize_url = make_url('oidc-authorize', params=params)

    if login_first:
        utils.login(app, simple_user)
    response = app.get(authorize_url)
    if not login_first:
        assert set(app.session['login-hint']) == {'backoffice', 'john@example.com'}
        response = response.follow()
        assert response.request.path == reverse('auth_login')
        response.form.set('username', simple_user.username)
        response.form.set('password', simple_user.username)
        response = response.form.submit(name='login-password-submit')
        response = response.follow()
        assert response.request.path == reverse('oidc-authorize')
    if oidc_client.authorization_mode != OIDCClient.AUTHORIZATION_MODE_NONE:
        response = response.maybe_follow()
        assert 'a2-oidc-authorization-form' in response.text
        assert OIDCAuthorization.objects.count() == 0
        assert OIDCCode.objects.count() == 0
        assert OIDCAccessToken.objects.count() == 0
        response.form['do_not_ask_again'] = do_not_ask_again
        response = response.form.submit('accept')
        if do_not_ask_again:
            assert OIDCAuthorization.objects.count() == 1
            authz = OIDCAuthorization.objects.get()
            assert authz.client == oidc_client
            assert authz.user == simple_user
            assert authz.scope_set() == set('openid profile email'.split())
            assert authz.expired >= now()
        else:
            assert OIDCAuthorization.objects.count() == 0
        utils.assert_event(
            'user.service.sso.authorization',
            session=app.session,
            user=simple_user,
            service=oidc_client,
            scopes=['email', 'openid', 'profile'],
        )
    utils.assert_event(
        'user.service.sso',
        session=app.session,
        user=simple_user,
        service=oidc_client,
        how='password-on-https',
    )
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        assert OIDCCode.objects.count() == 1
        code = OIDCCode.objects.get()
        assert code.client == oidc_client
        assert code.user == simple_user
        assert code.scope_set() == set('openid profile email'.split())
        assert code.state == 'xxx'
        assert code.nonce == 'yyy'
        assert code.redirect_uri == redirect_uri
        assert code.session_key == app.session.session_key
        assert code.auth_time <= now()
        assert code.expired >= now()
    assert response['Location'].startswith(redirect_uri)
    location = urllib.parse.urlparse(response['Location'])
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        query = urllib.parse.parse_qs(location.query)
        assert set(query.keys()) == {'code', 'state'}
        assert query['code'] == [code.uuid]
        code = query['code'][0]
        assert query['state'] == ['xxx']

        token_url = make_url('oidc-token')
        response = app.post(
            token_url,
            params={
                'grant_type': 'authorization_code',
                'code': code,
                'redirect_uri': oidc_client.redirect_uris.split()[0],
            },
            headers=client_authentication_headers(oidc_client),
        )
        assert 'error' not in response.json
        assert 'access_token' in response.json
        assert 'expires_in' in response.json
        assert 'id_token' in response.json
        assert response.json['token_type'] == 'Bearer'
        access_token = response.json['access_token']
        id_token = response.json['id_token']
    elif oidc_client.authorization_flow == oidc_client.FLOW_IMPLICIT:
        assert location.fragment
        query = urllib.parse.parse_qs(location.fragment)
        assert OIDCAccessToken.objects.count() == 1
        access_token = OIDCAccessToken.objects.get()
        assert set(query.keys()) == {'access_token', 'token_type', 'expires_in', 'id_token', 'state'}
        assert query['access_token'] == [access_token.uuid]
        assert query['token_type'] == ['Bearer']
        assert query['state'] == ['xxx']
        access_token = query['access_token'][0]
        id_token = query['id_token'][0]

    if oidc_client.idtoken_algo in (oidc_client.ALGO_RSA, oidc_client.ALGO_EC):
        key = JWKSet.from_json(app.get(reverse('oidc-certs')).content)
        algs = ['RS256', 'ES256']
    elif oidc_client.idtoken_algo == oidc_client.ALGO_HMAC:
        k = base64.b64encode(oidc_client.client_secret.encode('utf-8'))
        key = JWK(kty='oct', k=force_text(k))
        algs = ['HS256']
    else:
        raise NotImplementedError
    jwt = JWT(jwt=id_token, key=key, algs=algs)
    claims = json.loads(jwt.claims)
    assert set(claims) >= {'iss', 'sub', 'aud', 'exp', 'iat', 'nonce', 'auth_time', 'acr'}
    assert claims['nonce'] == 'yyy'
    assert response.request.url.startswith(claims['iss'])
    assert claims['aud'] == oidc_client.client_id
    assert parse_timestamp(claims['iat']) <= now()
    assert parse_timestamp(claims['auth_time']) <= now()
    exp_delta = (parse_timestamp(claims['exp']) - now()).total_seconds()
    assert exp_delta > 0
    if oidc_client.idtoken_duration:
        assert abs(exp_delta - oidc_client.idtoken_duration.total_seconds()) < 2
    else:
        assert abs(exp_delta - 30) < 2

    if login_first:
        assert claims['acr'] == '0'
    else:
        assert claims['acr'] == '1'
    assert claims['sub'] == make_sub(oidc_client, simple_user)
    assert claims['preferred_username'] == simple_user.username
    assert claims['given_name'] == simple_user.first_name
    assert claims['family_name'] == simple_user.last_name
    assert claims['email'] == simple_user.email
    assert claims['email_verified'] is False

    user_info_url = make_url('oidc-user-info')
    response = app.get(user_info_url, headers=bearer_authentication_headers(access_token))
    assert response.json['sub'] == make_sub(oidc_client, simple_user)
    assert response.json['preferred_username'] == simple_user.username
    assert response.json['given_name'] == simple_user.first_name
    assert response.json['family_name'] == simple_user.last_name
    assert response.json['email'] == simple_user.email
    assert response.json['email_verified'] is False

    # when adding extra attributes
    OIDCClaim.objects.create(client=oidc_client, name='ou', value='django_user_ou_name', scopes='profile')
    OIDCClaim.objects.create(client=oidc_client, name='roles', value='a2_role_names', scopes='profile, role')
    OIDCClaim.objects.create(
        client=oidc_client, name='cityscape_image', value='django_user_cityscape_image', scopes='profile'
    )
    simple_user.roles.add(
        get_role_model().objects.create(name='Whatever', slug='whatever', ou=get_default_ou())
    )
    response = app.get(user_info_url, headers=bearer_authentication_headers(access_token))
    assert response.json['ou'] == simple_user.ou.name
    assert response.json['roles'][0] == 'Whatever'
    assert response.json.get('cityscape_image') is None
    with open('tests/200x200.jpg', 'rb') as fd:
        simple_user.attributes.cityscape_image = File(fd)
    response = app.get(user_info_url, headers=bearer_authentication_headers(access_token))
    assert response.json['cityscape_image'].startswith('http://testserver/media/profile-image/')

    # check against a user without username
    simple_user.username = None
    simple_user.save()
    response = app.get(user_info_url, headers=bearer_authentication_headers(access_token))
    assert response.json['preferred_username'] == ''

    # Now logout
    if oidc_client.post_logout_redirect_uris:
        params = {
            'post_logout_redirect_uri': oidc_client.post_logout_redirect_uris,
            'state': 'xyz',
        }
        logout_url = make_url('oidc-logout', params=params)
        response = app.get(logout_url)
        assert response.status_code == 302
        assert response.location == 'https://example.com/?state=xyz'
        assert '_auth_user_id' not in app.session
    else:
        response = app.get(make_url('account_management'))
        response = response.click('Logout')
        if oidc_client.frontchannel_logout_uri:
            iframes = response.pyquery('iframe[src="https://example.com/southpark/logout/"]')
            assert iframes
            if oidc_client.frontchannel_timeout:
                assert iframes.attr('onload').endswith(', %d)' % oidc_client.frontchannel_timeout)
            else:
                assert iframes.attr('onload').endswith(', 300)')


def check_authorize_error(
    response, error, error_description, fragment, caplog, check_next=True, redirect_uri=None, message=True
):
    # check next_url qs
    if message:
        location = urllib.parse.urlparse(response.location)
        assert location.path == '/continue/'
        if check_next:
            location_qs = QueryDict(location.query or '')
            assert 'next' in location_qs
            assert location_qs['next'].startswith(redirect_uri)
            next_url = urllib.parse.urlparse(location_qs['next'])
            next_url_qs = QueryDict(next_url.fragment if fragment else next_url.query)
            assert next_url_qs['error'] == error
            assert next_url_qs['error_description'] == error_description
        # check continue page
        continue_response = response.follow()
        assert error_description in continue_response.pyquery('.error').text()
    elif check_next:
        assert response.location.startswith(redirect_uri)
        location = urllib.parse.urlparse(response.location)
        location_qs = QueryDict(location.fragment if fragment else location.query)
        assert location_qs['error'] == error
        assert location_qs['error_description'] == error_description
    # check logs
    last_record = caplog.records[-1]
    if message:
        assert last_record.levelname == 'WARNING'
    else:
        assert last_record.levelname == 'INFO'
    assert 'error "%s" in authorize endpoint' % error in last_record.message
    assert error_description in last_record.message
    if message:
        return continue_response


def assert_authorization_response(response, fragment=False, **kwargs):
    location = urllib.parse.urlparse(response.location)
    location_qs = QueryDict(location.fragment if fragment else location.query)
    assert set(location_qs) == set(kwargs)
    for key, value in kwargs.items():
        if value is None:
            assert key in location_qs
        elif isinstance(value, list):
            assert set(location_qs.getlist(key)) == set(value)
        else:
            assert value in location_qs[key]


@pytest.mark.parametrize('oidc_client', OIDC_CLIENT_PARAMS, indirect=True)
def test_invalid_request(oidc_client, caplog, oidc_settings, simple_user, app):
    redirect_uri = oidc_client.redirect_uris.split()[0]
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        fragment = False
        response_type = 'code'
    elif oidc_client.authorization_flow == oidc_client.FLOW_IMPLICIT:
        fragment = True
        response_type = 'id_token token'
    else:
        raise NotImplementedError

    assert_authorize_error = functools.partial(
        check_authorize_error, caplog=caplog, fragment=fragment, redirect_uri=redirect_uri
    )

    # missing client_id
    response = app.get(make_url('oidc-authorize', params={}))
    assert_authorize_error(response, 'invalid_request', 'Missing parameter "client_id"', check_next=False)

    # missing redirect_uri
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
            },
        )
    )
    assert_authorize_error(response, 'invalid_request', 'Missing parameter "redirect_uri"', check_next=False)

    # invalid client_id
    authorize_url = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': 'xxx',
                'redirect_uri': redirect_uri,
            },
        )
    )
    assert_authorize_error(response, 'invalid_request', 'Unknown client identifier: "xxx"', check_next=False)

    # invalid redirect_uri
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': 'xxx',
                'response_type': 'code',
                'scope': 'openid',
            },
        ),
        status=302,
    )
    continue_response = assert_authorize_error(
        response, 'invalid_request', 'Redirect URI "xxx" is unknown.', check_next=False
    )
    assert 'Known' not in continue_response.pyquery('.error').text()

    # invalid redirect_uri with DEBUG=True, list of redirect_uris is shown
    with override_settings(DEBUG=True):
        response = app.get(
            make_url(
                'oidc-authorize',
                params={
                    'client_id': oidc_client.client_id,
                    'redirect_uri': 'xxx',
                    'response_type': 'code',
                    'scope': 'openid',
                },
            ),
            status=302,
        )
        continue_response = assert_authorize_error(
            response, 'invalid_request', 'Redirect URI "xxx" is unknown.', check_next=False
        )
        assert (
            'Known redirect URIs are: https://example.com/callbac%C3%A9'
            in continue_response.pyquery('.error').text()
        )

    # missing response_type
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
            },
        )
    )
    assert_authorize_error(response, 'invalid_request', 'Missing parameter "response_type"')

    # unsupported response_type
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': 'xxx',
            },
        )
    )

    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        assert_authorize_error(response, 'unsupported_response_type', 'Response type must be "code"')
    elif oidc_client.authorization_flow == oidc_client.FLOW_IMPLICIT:
        assert_authorize_error(
            response, 'unsupported_response_type', 'Response type must be "id_token token" or "id_token"'
        )

    # missing scope
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
            },
        )
    )
    assert_authorize_error(response, 'invalid_request', 'Missing parameter "scope"')

    # invalid max_age : not an integer
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
                'scope': 'openid',
                'max_age': 'xxx',
            },
        )
    )
    assert_authorize_error(response, 'invalid_request', 'Parameter "max_age" must be a positive integer')

    # invalid max_age : not positive
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
                'scope': 'openid',
                'max_age': '-1',
            },
        )
    )
    assert_authorize_error(response, 'invalid_request', 'Parameter "max_age" must be a positive integer')

    # openid scope is missing
    authorize_url = make_url(
        'oidc-authorize',
        params={
            'client_id': oidc_client.client_id,
            'redirect_uri': redirect_uri,
            'response_type': response_type,
            'scope': 'profile',
        },
    )

    response = app.get(authorize_url)
    assert_authorize_error(response, 'invalid_scope', 'Scope must contain "openid", received "profile"')

    # use of an unknown scope
    authorize_url = make_url(
        'oidc-authorize',
        params={
            'client_id': oidc_client.client_id,
            'redirect_uri': redirect_uri,
            'response_type': response_type,
            'scope': 'openid email profile zob',
        },
    )

    response = app.get(authorize_url)
    assert_authorize_error(
        response,
        'invalid_scope',
        'Scope may contain "email, openid, profile" scope(s), received "email, openid, profile, zob"',
    )

    # restriction on scopes
    with override_settings(A2_IDP_OIDC_SCOPES=['openid']):
        response = app.get(
            make_url(
                'oidc-authorize',
                params={
                    'client_id': oidc_client.client_id,
                    'redirect_uri': redirect_uri,
                    'response_type': response_type,
                    'scope': 'openid email',
                },
            )
        )
        assert_authorize_error(
            response, 'invalid_scope', 'Scope may contain "openid" scope(s), received "email, openid"'
        )

    # cancel
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
                'scope': 'openid email profile',
                'cancel': '1',
            },
        )
    )
    assert_authorize_error(response, 'access_denied', 'Authentication cancelled by user', message=False)

    # prompt=none
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
                'scope': 'openid email profile',
                'prompt': 'none',
            },
        )
    )
    assert_authorize_error(
        response,
        'login_required',
        error_description='Login is required but prompt parameter is "none"',
        message=False,
    )

    utils.login(app, simple_user)

    # prompt=none max_age=0
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
                'scope': 'openid email profile',
                'max_age': '0',
                'prompt': 'none',
            },
        )
    )
    assert_authorize_error(
        response,
        'login_required',
        error_description='Login is required because of max_age, but prompt parameter is "none"',
        message=False,
    )

    # max_age=0
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
                'scope': 'openid email profile',
                'max_age': '0',
            },
        )
    )
    assert response.location.startswith(reverse('auth_login') + '?')

    # prompt=login
    authorize_url = make_url(
        'oidc-authorize',
        params={
            'client_id': oidc_client.client_id,
            'redirect_uri': redirect_uri,
            'response_type': response_type,
            'scope': 'openid email profile',
            'prompt': 'login',
        },
    )
    response = app.get(authorize_url)
    assert urllib.parse.urlparse(response['Location']).path == reverse('auth_login')

    if oidc_client.authorization_mode != oidc_client.AUTHORIZATION_MODE_NONE:
        # prompt is none, but consent is required
        response = app.get(
            make_url(
                'oidc-authorize',
                params={
                    'client_id': oidc_client.client_id,
                    'redirect_uri': redirect_uri,
                    'response_type': response_type,
                    'scope': 'openid email profile',
                    'prompt': 'none',
                },
            )
        )
        assert_authorize_error(
            response, 'consent_required', 'Consent is required but prompt parameter is "none"', message=False
        )

        # user do not consent
        response = app.get(
            make_url(
                'oidc-authorize',
                params={
                    'client_id': oidc_client.client_id,
                    'redirect_uri': redirect_uri,
                    'response_type': response_type,
                    'scope': 'openid email profile',
                },
            )
        )
        response = response.form.submit('refuse')
        assert_authorize_error(response, 'access_denied', 'User did not consent', message=False)

    # authorization exists
    authorize = OIDCAuthorization.objects.create(
        client=oidc_client,
        user=simple_user,
        scopes='openid profile email',
        expired=now() + datetime.timedelta(days=2),
    )
    response = app.get(
        make_url(
            'oidc-authorize',
            params={
                'client_id': oidc_client.client_id,
                'redirect_uri': redirect_uri,
                'response_type': response_type,
                'scope': 'openid email profile',
            },
        )
    )
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        assert_authorization_response(response, code=None)
    elif oidc_client.authorization_flow == oidc_client.FLOW_IMPLICIT:
        assert_authorization_response(
            response, access_token=None, id_token=None, expires_in=None, token_type=None, fragment=True
        )

    # client ask for explicit authorization
    authorize_url = make_url(
        'oidc-authorize',
        params={
            'client_id': oidc_client.client_id,
            'redirect_uri': redirect_uri,
            'response_type': response_type,
            'scope': 'openid email profile',
            'prompt': 'consent',
        },
    )
    response = app.get(authorize_url)
    assert 'a2-oidc-authorization-form' in response.text
    # check all authorization have been deleted, it's our policy
    assert OIDCAuthorization.objects.count() == 0
    if oidc_client.authorization_mode == oidc_client.AUTHORIZATION_MODE_NONE:
        # authorization mode is none, but explicit consent is asked, we validate it
        response = response.form.submit('accept')

    # authorization has expired
    OIDCCode.objects.all().delete()
    authorize.expired = now() - datetime.timedelta(days=2)
    authorize.save()
    response = app.get(authorize_url)
    assert 'a2-oidc-authorization-form' in response.text
    authorize.expired = now() + datetime.timedelta(days=2)
    authorize.scopes = 'openid profile'
    authorize.save()
    assert OIDCAuthorization.objects.count() == 1
    response.form['do_not_ask_again'] = True
    response = response.form.submit('accept')
    assert OIDCAuthorization.objects.count() == 1
    # old authorizations have been deleted
    assert OIDCAuthorization.objects.get().pk != authorize.pk

    # check expired codes
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        assert OIDCCode.objects.count() == 1
        code = OIDCCode.objects.get()
        assert code.is_valid()
        # make code expire
        code.expired = now() - datetime.timedelta(seconds=120)
        assert not code.is_valid()
        code.save()
        location = urllib.parse.urlparse(response['Location'])
        query = urllib.parse.parse_qs(location.query)
        assert set(query.keys()) == {'code'}
        assert query['code'] == [code.uuid]
        code = query['code'][0]
        token_url = make_url('oidc-token')
        params = {
            'grant_type': 'authorization_code',
            'redirect_uri': oidc_client.redirect_uris.split()[0],
        }
        response = app.post(
            token_url, params=params, headers=client_authentication_headers(oidc_client), status=400
        )
        assert response.json['error'] == 'invalid_request'
        assert response.json['error_description'] == 'Missing parameter "code"'

        params['code'] = code
        response = app.post(
            token_url, params=params, headers=client_authentication_headers(oidc_client), status=400
        )
        assert 'error' in response.json
        assert response.json['error'] == 'invalid_request'
        assert response.json['error_description'] == 'Parameter "code" has expired or user is disconnected'

    # invalid logout
    logout_url = make_url(
        'oidc-logout',
        params={
            'post_logout_redirect_uri': 'https://whatever.com/',
        },
    )
    response = app.get(logout_url)
    assert '_auth_user_id' in app.session
    assert 'Location' in response.headers

    # check code expiration after logout
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        code = OIDCCode.objects.get()
        code.expired = now() + datetime.timedelta(seconds=120)
        code.save()
        assert code.is_valid()
        utils.logout(app)
        code = OIDCCode.objects.get()
        assert not code.is_valid()
        response = app.post(
            token_url,
            params={
                'grant_type': 'authorization_code',
                'code': code.uuid,
                'redirect_uri': oidc_client.redirect_uris.split()[0],
            },
            headers=client_authentication_headers(oidc_client),
            status=400,
        )
        assert 'error' in response.json
        assert response.json['error'] == 'invalid_request'
        assert response.json['error_description'] == 'Parameter "code" has expired or user is disconnected'


def test_client_secret_post_authentication(oidc_settings, app, simple_oidc_client, simple_user):
    utils.login(app, simple_user)
    redirect_uri = simple_oidc_client.redirect_uris.split()[0]

    params = {
        'client_id': simple_oidc_client.client_id,
        'scope': 'openid profile email',
        'redirect_uri': redirect_uri,
        'state': 'xxx',
        'nonce': 'yyy',
        'response_type': 'code',
    }

    authorize_url = make_url('oidc-authorize', params=params)
    response = app.get(authorize_url)
    response = response.form.submit('accept')
    location = urllib.parse.urlparse(response['Location'])
    query = urllib.parse.parse_qs(location.query)
    code = query['code'][0]
    token_url = make_url('oidc-token')
    response = app.post(
        token_url,
        params={
            'grant_type': 'authorization_code',
            'code': code,
            'redirect_uri': redirect_uri,
            'client_id': simple_oidc_client.client_id,
            'client_secret': simple_oidc_client.client_secret,
        },
    )

    assert 'error' not in response.json
    assert 'access_token' in response.json
    assert 'expires_in' in response.json
    assert 'id_token' in response.json
    assert response.json['token_type'] == 'Bearer'


@pytest.mark.parametrize('login_first', [(True,), (False,)])
def test_role_control_access(login_first, oidc_settings, oidc_client, simple_user, app):
    # authorized_role
    role_authorized = get_role_model().objects.create(name='Goth Kids', slug='goth-kids', ou=get_default_ou())
    oidc_client.add_authorized_role(role_authorized)

    redirect_uri = oidc_client.redirect_uris.split()[0]
    params = {
        'client_id': oidc_client.client_id,
        'scope': 'openid profile email',
        'redirect_uri': redirect_uri,
        'state': 'xxx',
        'nonce': 'yyy',
    }

    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        params['response_type'] = 'code'
    elif oidc_client.authorization_flow == oidc_client.FLOW_IMPLICIT:
        params['response_type'] = 'token id_token'
    authorize_url = make_url('oidc-authorize', params=params)

    if login_first:
        utils.login(app, simple_user)

    # user not authorized
    response = app.get(authorize_url)
    assert 'https://example.com/southpark/' in response.text

    # user authorized
    simple_user.roles.add(role_authorized)
    simple_user.save()
    response = app.get(authorize_url)

    if not login_first:
        response = response.follow()
        response.form.set('username', simple_user.username)
        response.form.set('password', simple_user.username)
        response = response.form.submit(name='login-password-submit')
        response = response.follow()
    if oidc_client.authorization_mode != oidc_client.AUTHORIZATION_MODE_NONE:
        response.form['do_not_ask_again'] = True
        response = response.form.submit('accept')
        assert OIDCAuthorization.objects.get()
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        code = OIDCCode.objects.get()
    location = urllib.parse.urlparse(response['Location'])
    if oidc_client.authorization_flow == oidc_client.FLOW_AUTHORIZATION_CODE:
        query = urllib.parse.parse_qs(location.query)
        code = query['code'][0]
        token_url = make_url('oidc-token')
        response = app.post(
            token_url,
            params={
                'grant_type': 'authorization_code',
                'code': code,
                'redirect_uri': oidc_client.redirect_uris.split()[0],
            },
            headers=client_authentication_headers(oidc_client),
        )
        id_token = response.json['id_token']
    elif oidc_client.authorization_flow == oidc_client.FLOW_IMPLICIT:
        query = urllib.parse.parse_qs(location.fragment)
        id_token = query['id_token'][0]

    if oidc_client.idtoken_algo in (oidc_client.ALGO_RSA, oidc_client.ALGO_EC):
        key = JWKSet.from_json(app.get(reverse('oidc-certs')).content)
        algs = ['RS256', 'ES256']
    elif oidc_client.idtoken_algo == oidc_client.ALGO_HMAC:
        k = base64.b64encode(oidc_client.client_secret.encode('utf-8'))
        key = JWK(kty='oct', k=force_text(k))
        algs = ['HS256']
    else:
        raise NotImplementedError
    jwt = JWT(jwt=id_token, key=key, algs=algs)
    claims = json.loads(jwt.claims)
    if login_first:
        assert claims['acr'] == '0'
    else:
        assert claims['acr'] == '1'


def test_registration_service_slug(oidc_settings, app, simple_oidc_client, simple_user, hooks, mailoutbox):
    redirect_uri = simple_oidc_client.redirect_uris.split()[0]

    params = {
        'client_id': simple_oidc_client.client_id,
        'scope': 'openid profile email',
        'redirect_uri': redirect_uri,
        'state': 'xxx',
        'nonce': 'yyy',
        'response_type': 'code',
    }

    authorize_url = make_url('oidc-authorize', params=params)
    response = app.get(authorize_url)

    location = urllib.parse.urlparse(response['Location'])
    query = urllib.parse.parse_qs(location.query)
    assert query['service'] == ['default client']
    response = response.follow().click('Register')
    location = urllib.parse.urlparse(response.request.url)
    query = urllib.parse.parse_qs(location.query)
    assert query['service'] == ['default client']

    response.form.set('email', 'john.doe@example.com')
    response = response.form.submit()
    assert len(mailoutbox) == 1
    link = utils.get_link_from_mail(mailoutbox[0])
    response = app.get(link)
    response.form.set('first_name', 'John')
    response.form.set('last_name', 'Doe')
    response.form.set('password1', 'T0==toto')
    response.form.set('password2', 'T0==toto')
    response = response.form.submit()
    assert hooks.event[0]['kwargs']['name'] == 'sso-request'
    assert hooks.event[0]['kwargs']['service'].slug == 'client'

    assert hooks.event[1]['kwargs']['name'] == 'registration'
    assert hooks.event[1]['kwargs']['service'] == 'client'

    assert hooks.event[2]['kwargs']['name'] == 'login'
    assert hooks.event[2]['kwargs']['how'] == 'email'
    assert hooks.event[2]['kwargs']['service'] == 'client'


def test_claim_default_value(oidc_settings, normal_oidc_client, simple_user, app):
    oidc_settings.A2_IDP_OIDC_SCOPES = ['openid', 'profile', 'email', 'phone']
    Attribute.objects.create(
        name='phone',
        label='phone',
        kind='phone_number',
        asked_on_registration=False,
        required=False,
        user_visible=False,
        user_editable=False,
    )
    OIDCClaim.objects.create(
        client=normal_oidc_client, name='phone', value='django_user_phone', scopes='phone'
    )
    normal_oidc_client.authorization_flow = normal_oidc_client.FLOW_AUTHORIZATION_CODE
    normal_oidc_client.authorization_mode = normal_oidc_client.AUTHORIZATION_MODE_NONE
    normal_oidc_client.save()

    utils.login(app, simple_user)

    simple_user.username = None
    simple_user.save()

    oidc_client = normal_oidc_client
    redirect_uri = oidc_client.redirect_uris.split()[0]

    params = {
        'client_id': oidc_client.client_id,
        'scope': 'openid email profile phone',
        'redirect_uri': redirect_uri,
        'state': 'xxx',
        'nonce': 'yyy',
        'response_type': 'code',
    }

    def sso():
        authorize_url = make_url('oidc-authorize', params=params)

        response = app.get(authorize_url)
        location = urllib.parse.urlparse(response['Location'])
        query = urllib.parse.parse_qs(location.query)
        code = query['code'][0]

        token_url = make_url('oidc-token')
        response = app.post(
            token_url,
            params={
                'grant_type': 'authorization_code',
                'code': code,
                'redirect_uri': oidc_client.redirect_uris.split()[0],
            },
            headers=client_authentication_headers(oidc_client),
        )
        access_token = response.json['access_token']
        id_token = response.json['id_token']

        k = base64.b64encode(oidc_client.client_secret.encode('utf-8'))
        key = JWK(kty='oct', k=force_text(k))
        jwt = JWT(jwt=id_token, key=key)
        claims = json.loads(jwt.claims)

        user_info_url = make_url('oidc-user-info')
        response = app.get(user_info_url, headers=bearer_authentication_headers(access_token))
        return claims, response.json

    claims, user_info = sso()

    assert claims['sub'] == make_sub(oidc_client, simple_user)
    assert claims['preferred_username'] == ''
    assert claims['given_name'] == simple_user.first_name
    assert claims['family_name'] == simple_user.last_name
    assert claims['email'] == simple_user.email
    assert claims['phone'] is None
    assert claims['email_verified'] is False

    assert user_info['sub'] == make_sub(oidc_client, simple_user)
    assert user_info['preferred_username'] == ''
    assert user_info['given_name'] == simple_user.first_name
    assert user_info['family_name'] == simple_user.last_name
    assert user_info['email'] == simple_user.email
    assert user_info['phone'] is None
    assert user_info['email_verified'] is False

    params['scope'] = 'openid email'

    claims, user_info = sso()

    assert claims['sub'] == make_sub(oidc_client, simple_user)
    assert claims['email'] == simple_user.email
    assert claims['email_verified'] is False
    assert 'phone' not in claims
    assert 'preferred_username' not in claims
    assert 'given_name' not in claims
    assert 'family_name' not in claims

    assert user_info['sub'] == make_sub(oidc_client, simple_user)
    assert user_info['email'] == simple_user.email
    assert user_info['email_verified'] is False
    assert 'phone' not in user_info
    assert 'preferred_username' not in user_info
    assert 'given_name' not in user_info
    assert 'family_name' not in user_info


def test_claim_templated(oidc_settings, normal_oidc_client, simple_user, app):
    oidc_settings.A2_IDP_OIDC_SCOPES = ['openid', 'profile', 'email']
    OIDCClaim.objects.filter(client=normal_oidc_client, name='given_name').delete()
    OIDCClaim.objects.filter(client=normal_oidc_client, name='family_name').delete()
    OIDCClaim.objects.create(
        client=normal_oidc_client,
        name='given_name',
        value='{{ django_user_first_name|add:"ounet" }}',
        scopes='profile',
    )
    OIDCClaim.objects.create(
        client=normal_oidc_client,
        name='family_name',
        value='{{ "Von der "|add:django_user_last_name }}',
        scopes='profile',
    )
    normal_oidc_client.authorization_flow = normal_oidc_client.FLOW_AUTHORIZATION_CODE
    normal_oidc_client.authorization_mode = normal_oidc_client.AUTHORIZATION_MODE_NONE
    normal_oidc_client.save()

    utils.login(app, simple_user)

    oidc_client = normal_oidc_client
    redirect_uri = oidc_client.redirect_uris.split()[0]

    params = {
        'client_id': oidc_client.client_id,
        'scope': 'openid email profile',
        'redirect_uri': redirect_uri,
        'state': 'xxx',
        'nonce': 'yyy',
        'response_type': 'code',
    }

    def sso():
        authorize_url = make_url('oidc-authorize', params=params)

        response = app.get(authorize_url)
        location = urllib.parse.urlparse(response['Location'])
        query = urllib.parse.parse_qs(location.query)
        code = query['code'][0]

        token_url = make_url('oidc-token')
        response = app.post(
            token_url,
            params={
                'grant_type': 'authorization_code',
                'code': code,
                'redirect_uri': oidc_client.redirect_uris.split()[0],
            },
            headers=client_authentication_headers(oidc_client),
        )
        access_token = response.json['access_token']
        id_token = response.json['id_token']

        k = base64.b64encode(oidc_client.client_secret.encode('utf-8'))
        key = JWK(kty='oct', k=force_text(k))
        jwt = JWT(jwt=id_token, key=key)
        claims = json.loads(jwt.claims)

        user_info_url = make_url('oidc-user-info')
        response = app.get(user_info_url, headers=bearer_authentication_headers(access_token))
        return claims, response.json

    claims, user_info = sso()

    assert claims['given_name'].endswith('ounet')
    assert claims['given_name'].startswith(simple_user.first_name)
    assert claims['family_name'].startswith('Von der')
    assert claims['family_name'].endswith(simple_user.last_name)

    assert user_info['given_name'].endswith('ounet')
    assert user_info['given_name'].startswith(simple_user.first_name)
    assert user_info['family_name'].startswith('Von der')
    assert user_info['family_name'].endswith(simple_user.last_name)


def test_client_validate_redirect_uri():
    client = OIDCClient(
        redirect_uris='''http://example.com
http://example2.com/
http://example3.com/toto
http://*example4.com/
http://example5.com/toto*
http://example6.com/#*
http://example7.com/?*
http://example8.com/?*#*
'''
    )
    # ok
    for uri in [
        'http://example.com',
        'http://example.com/',
        'http://example2.com',
        'http://example2.com/',
        'http://example3.com/toto',
        'http://example3.com/toto/',
        'http://example4.com/',
        'http://example4.com',
        'http://coin.example4.com',
        'http://coin.example4.com/',
        'http://example5.com/toto',
        'http://example5.com/toto/',
        'http://example5.com/toto/tata',
        'http://example5.com/toto/tata/',
        'http://example6.com/#some-fragment',
        'http://example7.com/?foo=bar',
        'http://example8.com/?foo=bar#some-fragment',
    ]:
        client.validate_redirect_uri(uri)
    # nok
    for uri in [
        'http://coin.example.com/',
        'http://example.com/toto/',
        'http://coin.example.com',
        'http://example3.com/',
        'http://example3.com',
        'http://coinexample4.com',
        'http://coinexample4.com/',
        'http://example5.com/tototata/',
        'http://example5.com/tototata',
        'http://example6.com/?foo=bar',
        'http://example7.com/#some-fragment',
    ]:
        with pytest.raises(ValueError, match=r'is not declared'):
            client.validate_redirect_uri(uri)
    client.validate_redirect_uri('http://example5.com/toto/' + 'a' * 500)
    with pytest.raises(ValueError, match=r'redirect_uri length >'):
        client.validate_redirect_uri('http://example5.com/toto/' + 'a' * 1024)


def test_filter_api_users(app, oidc_client, admin, simple_user, role_random):
    oidc_client.has_api_access = True
    oidc_client.save()

    if oidc_client.identifier_policy not in (oidc_client.POLICY_UUID, oidc_client.POLICY_PAIRWISE_REVERSIBLE):
        return

    app.authorization = ('Basic', (oidc_client.client_id, oidc_client.client_secret))

    response = app.get('/api/users/')
    count = len(response.json['results'])
    assert count > 0

    AuthorizedRole.objects.create(service=oidc_client, role=role_random)

    response = app.get('/api/users/')
    assert len(response.json['results']) == 0

    role_random.members.add(simple_user)
    response = app.get('/api/users/')
    assert len(response.json['results']) == 1

    AuthorizedRole.objects.all().delete()

    response = app.get('/api/users/')
    assert len(response.json['results']) == count


def test_credentials_grant(app, oidc_client, admin, simple_user):
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.scope = 'openid'
    oidc_client.save()
    token_url = make_url('oidc-token')
    if oidc_client.idtoken_algo == OIDCClient.ALGO_HMAC:
        k = base64url(oidc_client.client_secret.encode('utf-8'))
        jwk = JWK(kty='oct', k=force_text(k))
    elif oidc_client.idtoken_algo == OIDCClient.ALGO_RSA:
        jwk = get_first_rsa_sig_key()
    elif oidc_client.idtoken_algo == OIDCClient.ALGO_EC:
        jwk = get_first_ec_sig_key()

    # 1. test in-request client credentials
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'username': simple_user.username,
        'password': simple_user.username,
    }
    response = app.post(token_url, params=params)
    assert 'id_token' in response.json
    token = response.json['id_token']
    header, payload, signature = token.split('.')
    jwt = JWT()
    # jwt deserialization implicitly checks the token signature:
    jwt.deserialize(token, key=jwk)
    claims = json.loads(jwt.claims)
    assert set(claims) == {'acr', 'aud', 'auth_time', 'exp', 'iat', 'iss', 'sub'}
    assert all(claims.values())

    # 2. test basic authz
    params.pop('client_id')
    params.pop('client_secret')

    response = app.post(token_url, params=params, headers=client_authentication_headers(oidc_client))
    assert 'id_token' in response.json
    token = response.json['id_token']
    header, payload, signature = token.split('.')
    jwt = JWT()
    # jwt deserialization implicitly checks the token signature:
    jwt.deserialize(token, key=jwk)
    claims = json.loads(jwt.claims)
    assert set(claims) == {'acr', 'aud', 'auth_time', 'exp', 'iat', 'iss', 'sub'}
    assert all(claims.values())


def test_credentials_grant_ratelimitation_invalid_client(
    app, oidc_client, admin, simple_user, oidc_settings, freezer
):
    freezer.move_to('2020-01-01')

    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    token_url = make_url('oidc-token')
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': 'notgood',
        'grant_type': 'password',
        'username': simple_user.username,
        'password': simple_user.username,
    }
    for i in range(int(oidc_settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT.split('/')[0])):
        response = app.post(token_url, params=params, status=400)
        assert response.json['error'] == 'invalid_client'
        assert 'Wrong client secret' in response.json['error_description']
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'invalid_request'
    assert response.json['error_description'] == 'Rate limit exceeded for IP address "127.0.0.1"'


def test_credentials_grant_ratelimitation_valid_client(
    app, oidc_client, admin, simple_user, oidc_settings, freezer
):
    freezer.move_to('2020-01-01')

    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    token_url = make_url('oidc-token')
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'username': simple_user.username,
        'password': simple_user.username,
    }
    for i in range(int(oidc_settings.A2_IDP_OIDC_PASSWORD_GRANT_RATELIMIT.split('/')[0])):
        app.post(token_url, params=params)
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'invalid_client'
    assert response.json['error_description'] == 'Rate limit of 100/m exceeded for client "oidcclient"'


def test_credentials_grant_retrytimout(app, oidc_client, admin, simple_user, settings, freezer):
    freezer.move_to('2020-01-01')

    settings.A2_LOGIN_EXPONENTIAL_RETRY_TIMEOUT_DURATION = 2
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    token_url = make_url('oidc-token')
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'username': simple_user.username,
        'password': 'SurelyNotTheRightPassword',
    }
    attempts = 0
    while attempts < 100:
        response = app.post(token_url, params=params, status=400)
        attempts += 1
        if attempts >= 10:
            assert response.json['error'] == 'invalid_request'
            assert 'Too many attempts with erroneous RO password' in response.json['error_description']

    # freeze some time after backoff delay expiration
    freezer.move_to(datetime.timedelta(days=2))

    # obtain a successful login
    params['password'] = simple_user.username
    response = app.post(token_url, params=params, status=200)
    assert 'id_token' in response.json


def test_credentials_grant_invalid_flow(app, oidc_client, admin, simple_user, settings):
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'username': simple_user.username,
        'password': 'SurelyNotTheRightPassword',
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'unauthorized_client'
    assert 'is not configured' in response.json['error_description']


def test_credentials_grant_invalid_client(app, oidc_client, admin, simple_user, settings):
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': 'tryingthis',  # Nope, wrong secret
        'grant_type': 'password',
        'username': simple_user.username,
        'password': simple_user.username,
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'invalid_client'
    assert response.json['error_description'] == 'Wrong client secret'


def test_credentials_grant_unauthz_client(app, oidc_client, admin, simple_user, settings):
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'username': simple_user.username,
        'password': simple_user.username,
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'unauthorized_client'
    assert 'Client is not configured for resource owner' in response.json['error_description']


def test_credentials_grant_invalid_content_type(app, oidc_client, admin, simple_user, settings):
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'username': simple_user.username,
        'password': simple_user.username,
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params, content_type='multipart/form-data', status=400)
    assert response.json['error'] == 'invalid_request'
    assert 'Wrong content type' in response.json['error_description']


def test_credentials_grant_ou_selection_simple(
    app, oidc_client, admin, user_ou1, user_ou2, ou1, ou2, settings
):
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'ou_slug': ou1.slug,
        'username': user_ou1.username,
        'password': user_ou1.username,
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params, status=200)

    params['username'] = user_ou2.username
    params['password'] = user_ou2.password
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'access_denied'
    assert response.json['error_description'] == 'Invalid user credentials'


def test_credentials_grant_ou_selection_username_not_unique(
    app, oidc_client, admin, user_ou1, admin_ou2, ou1, ou2, settings
):
    settings.A2_USERNAME_IS_UNIQUE = False
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    admin_ou2.username = user_ou1.username
    admin_ou2.set_password(user_ou1.username)
    admin_ou2.save()
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'ou_slug': ou1.slug,
        'username': user_ou1.username,
        'password': user_ou1.username,
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params)
    assert OIDCAccessToken.objects.get(uuid=response.json['access_token']).user == user_ou1

    params['ou_slug'] = ou2.slug
    response = app.post(token_url, params=params)
    assert OIDCAccessToken.objects.get(uuid=response.json['access_token']).user == admin_ou2


def test_credentials_grant_ou_selection_username_not_unique_wrong_ou(
    app, oidc_client, admin, user_ou1, admin_ou2, ou1, ou2, settings
):
    settings.A2_USERNAME_IS_UNIQUE = False
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'ou_slug': ou2.slug,
        'username': user_ou1.username,
        'password': user_ou1.username,
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params, status=400)

    params['ou_slug'] = ou1.slug
    params['username'] = admin_ou2.username
    params['password'] = admin_ou2.password
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'access_denied'
    assert response.json['error_description'] == 'Invalid user credentials'


def test_credentials_grant_ou_selection_invalid_ou(app, oidc_client, admin, user_ou1, settings):
    oidc_client.authorization_flow = OIDCClient.FLOW_RESOURCE_OWNER_CRED
    oidc_client.save()
    params = {
        'client_id': oidc_client.client_id,
        'client_secret': oidc_client.client_secret,
        'grant_type': 'password',
        'ou_slug': 'invalidslug',
        'username': user_ou1.username,
        'password': user_ou1.username,
    }
    token_url = make_url('oidc-token')
    response = app.post(token_url, params=params, status=400)
    assert response.json['error'] == 'invalid_request'
    assert (
        response.json['error_description']
        == 'Parameter "ou_slug" does not match an existing organizational unit'
    )


def test_oidc_client_clean():
    OIDCClient(
        redirect_uris='https://example.com/ https://example2.com/', identifier_policy=OIDCClient.POLICY_UUID
    ).clean()

    with pytest.raises(ValidationError, match=r'same domain'):
        OIDCClient(
            redirect_uris='https://example.com/ https://example2.com/',
            identifier_policy=OIDCClient.POLICY_PAIRWISE_REVERSIBLE,
        ).clean()

    with pytest.raises(ValidationError, match=r'same domain'):
        OIDCClient(
            redirect_uris='https://example.com/ https://example2.com/',
            identifier_policy=OIDCClient.POLICY_PAIRWISE,
        ).clean()

    with pytest.raises(ValidationError, match=r'same domain'):
        OIDCClient(
            redirect_uris='https://example.com/ https://example2.com/',
            identifier_policy=OIDCClient.POLICY_PAIRWISE,
        ).clean()

    with pytest.raises(ValidationError, match=r'within an OU'):
        OIDCClient(authorization_mode=OIDCClient.AUTHORIZATION_MODE_BY_OU, ou=None).clean()

    OIDCClient(
        redirect_uris='https://example.com/ https://example2.com/',
        sector_identifier_uri='https://example.com/',
    ).clean()


def test_oidc_authorized_oauth_services_view(app, oidc_client, simple_user):
    from django.contrib.contenttypes.models import ContentType

    url = make_url('authorized-oauth-services')
    response = app.get(url, status=302)
    assert '/login/' in response.location

    utils.login(app, simple_user)
    response = app.get(url, status=200)
    assert "You have not given any authorization to access your account profile data." in response.text

    # create an ou authz
    OU = get_ou_model()
    ou1 = OU.objects.create(name='Orgunit1', slug='orgunit1')
    OIDCAuthorization.objects.create(
        client=ou1,
        user=simple_user,
        scopes='openid profile email',
        expired=now() + datetime.timedelta(days=2),
    )
    # create service authzs
    OIDCAuthorization.objects.create(
        client=oidc_client, user=simple_user, scopes='openid', expired=now() + datetime.timedelta(days=2)
    )
    OIDCAuthorization.objects.create(
        client=oidc_client,
        user=simple_user,
        scopes='openid profile',
        expired=now() + datetime.timedelta(days=2),
    )
    OIDCAuthorization.objects.create(
        client=oidc_client,
        user=simple_user,
        scopes='openid profile email',
        expired=now() + datetime.timedelta(days=2),
    )

    response = app.get(url, status=200)
    assert "You have given authorizations to access your account profile data." in response.text
    assert len(response.html.find_all('button', {'class': 'authorized-oauth-services--revoke-button'})) == 4

    # revoke two service authz
    response = response.forms[1].submit()
    response = response.follow()
    assert len(response.html.find_all('button', {'class': 'authorized-oauth-services--revoke-button'})) == 3
    assert (
        OIDCAuthorization.objects.filter(client_ct=ContentType.objects.get_for_model(OIDCClient)).count() == 2
    )
    response = response.forms[1].submit()
    response = response.follow()
    assert len(response.html.find_all('button', {'class': 'authorized-oauth-services--revoke-button'})) == 2
    assert (
        OIDCAuthorization.objects.filter(client_ct=ContentType.objects.get_for_model(OIDCClient)).count() == 1
    )

    # revoke the only OU authz
    response = response.forms[0].submit()
    response = response.follow()
    assert len(response.html.find_all('button', {'class': 'authorized-oauth-services--revoke-button'})) == 1
    assert OIDCAuthorization.objects.filter(client_ct=ContentType.objects.get_for_model(OU)).count() == 0


def test_oidc_good_next_url_hook(app, oidc_client):
    from django.test.client import RequestFactory

    rf = RequestFactory()
    request = rf.get('/')
    assert good_next_url(request, 'https://example.com/')
