<filename>hydra_login2f/hydra.py
from urllib.parse import urljoin, quote_plus
import requests
from flask import current_app
from .redis import increment_key_with_limit, UserLoginsHistory, ExceededValueLimitError


def get_subject(user_id):
    return current_app.config['SUBJECT_PREFIX'] + str(user_id)


def invalidate_credentials(user_id):
    UserLoginsHistory(user_id).clear()
    subject = quote_plus(get_subject(user_id))
    timeout = current_app.config['HYDRA_REQUEST_TIMEOUT_SECONDS']
    hydra_consents_base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/sessions/consent')
    hydra_logins_base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/sessions/login')
    requests.delete(f'{hydra_consents_base_url}?subject={subject}', timeout=timeout)
    requests.delete(f'{hydra_logins_base_url}?subject={subject}', timeout=timeout)


class LoginRequest:
    LOGIN_COUNT_SUBJECT_PREFIX = 'logins:'

    class TooManyLogins(Exception):
        """Too many login attempts."""

    def __init__(self, challenge_id):
        self.challenge_id = quote_plus(challenge_id)
        self.timeout = current_app.config['HYDRA_REQUEST_TIMEOUT_SECONDS']
        base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/requests/')
        self.fetch_url = urljoin(base_url, 'login')
        self.accept_url = urljoin(base_url, 'login/accept')
        self.reject_url = urljoin(base_url, 'login/reject')

    def register_successful_login(self, subject):
        key = self.LOGIN_COUNT_SUBJECT_PREFIX + subject
        try:
            increment_key_with_limit(key, limit=current_app.config['MAX_LOGINS_PER_MONTH'], period_seconds=2600000)
        except ExceededValueLimitError:
            raise self.TooManyLogins()

    def fetch(self):
        """Return the subject if already logged, `None` otherwise."""

        r = requests.get(f'{self.fetch_url}?login_challenge={self.challenge_id}', timeout=self.timeout)
        r.raise_for_status()
        fetched_data = r.json()
        return fetched_data['subject'] if fetched_data['skip'] else None

    def accept(self, subject, remember=False, remember_for=1000000000):
        """Accept the request unless the limit is reached, return an URL to redirect to."""

        try:
            self.register_successful_login(subject)
        except self.TooManyLogins:
            return self.reject()
        r = requests.put(f'{self.accept_url}?login_challenge={self.challenge_id}', timeout=self.timeout, json={
            'subject': subject,
            'remember': remember,
            'remember_for': remember_for,
        })
        r.raise_for_status()
        return r.json()['redirect_to']

    def reject(self):
        """Reject the request, return an URL to redirect to."""

        r = requests.put(f'{self.reject_url}?login_challenge={self.challenge_id}', timeout=self.timeout, json={
            'error': 'too_many_logins',
            'error_description': 'Too many login attempts have been made in a given period of time.',
            'error_hint': 'Try again later.',
        })
        r.raise_for_status()
        return r.json()['redirect_to']


class ConsentRequest:
    def __init__(self, challenge_id):
        self.challenge_id = quote_plus(challenge_id)
        self.timeout = current_app.config['HYDRA_REQUEST_TIMEOUT_SECONDS']
        base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/requests/')
        self.fetch_url = urljoin(base_url, 'consent')
        self.accept_url = urljoin(base_url, 'consent/accept')

    def fetch(self):
        """Return the list of requested scopes, or an empty list if no consent is required."""

        r = requests.get(f'{self.fetch_url}?consent_challenge={self.challenge_id}', timeout=self.timeout)
        r.raise_for_status()
        fetched_data = r.json()
        return [] if fetched_data['skip'] else fetched_data['requested_scope']

    def accept(self, grant_scope, remember=False, remember_for=0):
        """Approve the request, return an URL to redirect to."""

        r = requests.put(f'{self.accept_url}?consent_challenge={self.challenge_id}', timeout=self.timeout, json={
            'grant_scope': grant_scope,
            'remember': remember,
            'remember_for': remember_for,
        })
        r.raise_for_status()
        return r.json()['redirect_to']

<filename>hydra_login2f/hydra.py
from urllib.parse import urljoin, quote_plus
import requests
from flask import current_app
from .redis import increment_key_with_limit, UserLoginsHistory, ExceededValueLimitError


def get_subject(user_id):
    return current_app.config['SUBJECT_PREFIX'] + str(user_id)


def invalidate_credentials(user_id):
    UserLoginsHistory(user_id).clear()
    subject = quote_plus(get_subject(user_id))
    timeout = current_app.config['HYDRA_REQUEST_TIMEOUT_SECONDS']
    hydra_consents_base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/sessions/consent')
    hydra_logins_base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/sessions/login')
    requests.delete(f'{hydra_consents_base_url}?subject={subject}', timeout=timeout)
    requests.delete(f'{hydra_logins_base_url}?subject={subject}', timeout=timeout)


class LoginRequest:
    LOGIN_COUNT_SUBJECT_PREFIX = 'logins:'

    class TooManyLogins(Exception):
        """Too many login attempts."""

    def __init__(self, challenge_id):
        self.challenge_id = quote_plus(challenge_id)
        self.timeout = current_app.config['HYDRA_REQUEST_TIMEOUT_SECONDS']
        base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/requests/')
        self.fetch_url = urljoin(base_url, 'login')
        self.accept_url = urljoin(base_url, 'login/accept')
        self.reject_url = urljoin(base_url, 'login/reject')

    def register_successful_login(self, subject):
        key = self.LOGIN_COUNT_SUBJECT_PREFIX + subject
        try:
            increment_key_with_limit(key, limit=current_app.config['MAX_LOGINS_PER_MONTH'], period_seconds=2600000)
        except ExceededValueLimitError:
            raise self.TooManyLogins()

    def fetch(self):
        """Return the subject if already logged, `None` otherwise."""

        r = requests.get(f'{self.fetch_url}?login_challenge={self.challenge_id}', timeout=self.timeout)
        r.raise_for_status()
        fetched_data = r.json()
        return fetched_data['subject'] if fetched_data['skip'] else None

    def accept(self, subject, remember=False, remember_for=1000000000):
        """Accept the request unless the limit is reached, return an URL to redirect to."""

        try:
            self.register_successful_login(subject)
        except self.TooManyLogins:
            return self.reject()
        r = requests.put(f'{self.accept_url}?login_challenge={self.challenge_id}', timeout=self.timeout, json={
            'subject': subject,
            'remember': remember,
            'remember_for': remember_for,
        })
        r.raise_for_status()
        return r.json()['redirect_to']

    def reject(self):
        """Reject the request, return an URL to redirect to."""

        r = requests.put(f'{self.reject_url}?login_challenge={self.challenge_id}', timeout=self.timeout, json={
            'error': 'too_many_logins',
            'error_description': 'Too many login attempts have been made in a given period of time.',
            'error_hint': 'Try again later.',
        })
        r.raise_for_status()
        return r.json()['redirect_to']


class ConsentRequest:
    def __init__(self, challenge_id):
        self.challenge_id = quote_plus(challenge_id)
        self.timeout = current_app.config['HYDRA_REQUEST_TIMEOUT_SECONDS']
        base_url = urljoin(current_app.config['HYDRA_ADMIN_URL'], '/oauth2/auth/requests/')
        self.fetch_url = urljoin(base_url, 'consent')
        self.accept_url = urljoin(base_url, 'consent/accept')

    def fetch(self):
        """Return the list of requested scopes, or an empty list if no consent is required."""

        r = requests.get(f'{self.fetch_url}?consent_challenge={self.challenge_id}', timeout=self.timeout)
        r.raise_for_status()
        fetched_data = r.json()
        return [] if fetched_data['skip'] else fetched_data['requested_scope']

    def accept(self, grant_scope, remember=False, remember_for=0):
        """Approve the request, return an URL to redirect to."""

        r = requests.put(f'{self.accept_url}?consent_challenge={self.challenge_id}', timeout=self.timeout, json={
            'grant_scope': grant_scope,
            'remember': remember,
            'remember_for': remember_for,
        })
        r.raise_for_status()
        return r.json()['redirect_to']

Too many login attempts. Return the subject if already logged, `None` otherwise. Accept the request unless the limit is reached, return an URL to redirect to. Reject the request, return an URL to redirect to. Return the list of requested scopes, or an empty list if no consent is required. Approve the request, return an URL to redirect to.