quay/data/users/__init__.py

import itertools
import json
import logging
import uuid
from typing import Union

import features
from data import model
from data.users.apptoken import AppTokenInternalAuth
from data.users.database import DatabaseUsers
from data.users.externaljwt import ExternalJWTAuthN
from data.users.externalldap import LDAPUsers
from data.users.externaloidc import OIDCUsers
from data.users.federated import FederatedUsers
from data.users.keystone import get_keystone_users
from oauth.oidc import OIDCLoginService
from util.config.superusermanager import ConfigUserManager
from util.security.aes import AESCipher
from util.security.secret import convert_secret_key

logger = logging.getLogger(__name__)


def get_federated_service_name(authentication_type):
    if authentication_type == "LDAP":
        return "ldap"

    if authentication_type == "JWT":
        return "jwtauthn"

    if authentication_type == "Keystone":
        return "keystone"

    if authentication_type == "AppToken":
        return None

    if authentication_type == "Database":
        return None

    if authentication_type == "OIDC":
        return "oidc"

    raise Exception("Unknown auth type: %s" % authentication_type)


LDAP_CERT_FILENAME = "ldap.crt"


def get_users_handler(config, _, override_config_dir, oauth_login):
    """
    Returns a users handler for the authentication configured in the given config object.
    """
    authentication_type = config.get("AUTHENTICATION_TYPE", "Database")

    if authentication_type == "Database":
        return DatabaseUsers()

    if authentication_type == "LDAP":
        ldap_uri = config.get("LDAP_URI", "ldap://localhost")
        base_dn = config.get("LDAP_BASE_DN")
        admin_dn = config.get("LDAP_ADMIN_DN")
        admin_passwd = config.get("LDAP_ADMIN_PASSWD")
        user_rdn = config.get("LDAP_USER_RDN", [])
        uid_attr = config.get("LDAP_UID_ATTR", "uid")
        email_attr = config.get("LDAP_EMAIL_ATTR", "mail")
        memberof_attr = config.get("LDAP_MEMBEROF_ATTR", "memberOf")
        secondary_user_rdns = config.get("LDAP_SECONDARY_USER_RDNS", [])
        timeout = config.get("LDAP_TIMEOUT")
        network_timeout = config.get("LDAP_NETWORK_TIMEOUT")
        ldap_user_filter = config.get("LDAP_USER_FILTER", None)
        ldap_superuser_filter = config.get("LDAP_SUPERUSER_FILTER", None)
        ldap_global_readonly_superuser_filter = config.get(
            "LDAP_GLOBAL_READONLY_SUPERUSER_FILTER", None
        )
        ldap_restricted_user_filter = config.get("LDAP_RESTRICTED_USER_FILTER", None)
        ldap_referrals = int(config.get("LDAP_FOLLOW_REFERRALS", True))

        allow_tls_fallback = config.get("LDAP_ALLOW_INSECURE_FALLBACK", False)
        return LDAPUsers(
            ldap_uri,
            base_dn,
            admin_dn,
            admin_passwd,
            user_rdn,
            uid_attr,
            email_attr,
            memberof_attr,
            allow_tls_fallback,
            secondary_user_rdns=secondary_user_rdns,
            requires_email=features.MAILING,
            timeout=timeout,
            network_timeout=network_timeout,
            ldap_user_filter=ldap_user_filter,
            ldap_superuser_filter=ldap_superuser_filter,
            ldap_global_readonly_superuser_filter=ldap_global_readonly_superuser_filter,
            ldap_restricted_user_filter=ldap_restricted_user_filter,
            ldap_referrals=ldap_referrals,
        )

    if authentication_type == "JWT":
        verify_url = config.get("JWT_VERIFY_ENDPOINT")
        issuer = config.get("JWT_AUTH_ISSUER")
        max_fresh_s = config.get("JWT_AUTH_MAX_FRESH_S", 300)

        query_url = config.get("JWT_QUERY_ENDPOINT", None)
        getuser_url = config.get("JWT_GETUSER_ENDPOINT", None)

        return ExternalJWTAuthN(
            verify_url,
            query_url,
            getuser_url,
            issuer,
            override_config_dir,
            config["HTTPCLIENT"],
            max_fresh_s,
            requires_email=features.MAILING,
        )

    if authentication_type == "Keystone":
        auth_url = config.get("KEYSTONE_AUTH_URL")
        auth_version = int(config.get("KEYSTONE_AUTH_VERSION", 2))
        timeout = config.get("KEYSTONE_AUTH_TIMEOUT")
        keystone_admin_username = config.get("KEYSTONE_ADMIN_USERNAME")
        keystone_admin_password = config.get("KEYSTONE_ADMIN_PASSWORD")
        keystone_admin_tenant = config.get("KEYSTONE_ADMIN_TENANT")
        return get_keystone_users(
            auth_version,
            auth_url,
            keystone_admin_username,
            keystone_admin_password,
            keystone_admin_tenant,
            timeout,
            requires_email=features.MAILING,
        )

    if authentication_type == "AppToken":
        if features.DIRECT_LOGIN:
            raise Exception("Direct login feature must be disabled to use AppToken internal auth")

        if not features.APP_SPECIFIC_TOKENS:
            raise Exception(
                "AppToken internal auth requires app specific token support to be enabled"
            )

        return AppTokenInternalAuth()

    if authentication_type == "OIDC" and oauth_login:
        for service in oauth_login.services:
            if isinstance(service, OIDCLoginService):
                config = service.config
                client_id = config.get("CLIENT_ID", None)
                client_secret = config.get("CLIENT_SECRET", None)
                oidc_server = config.get("OIDC_SERVER", None)
                service_name = config.get("SERVICE_NAME", None)
                login_scopes = config.get("LOGIN_SCOPES", None)
                preferred_group_claim_name = config.get("PREFERRED_GROUP_CLAIM_NAME", None)

                return OIDCUsers(
                    client_id,
                    client_secret,
                    oidc_server,
                    service_name,
                    login_scopes,
                    preferred_group_claim_name,
                )

    raise RuntimeError("Unknown authentication type: %s" % authentication_type)


class UserAuthentication(object):
    def __init__(self, app=None, config_provider=None, override_config_dir=None, oauth_login=None):
        self.secret_key = None
        self.app = app
        if app is not None:
            self.state = self.init_app(app, config_provider, override_config_dir, oauth_login)
        else:
            self.state = None

    def init_app(self, app, config_provider, override_config_dir, oauth_login):
        self.secret_key = convert_secret_key(app.config["SECRET_KEY"])
        users = get_users_handler(app.config, config_provider, override_config_dir, oauth_login)

        # register extension with app
        app.extensions = getattr(app, "extensions", {})
        app.extensions["authentication"] = users

        return users

    def encrypt_user_password(self, password):
        """
        Returns an encrypted version of the user's password.
        """
        data = {"password": password}

        message = json.dumps(data).encode("utf-8")
        cipher = AESCipher(self.secret_key)
        return cipher.encrypt(message)

    def _decrypt_user_password(self, encrypted):
        """
        Attempts to decrypt the given password and returns it.
        """
        cipher = AESCipher(self.secret_key)

        try:
            message = cipher.decrypt(encrypted)
        except ValueError:
            return None
        except TypeError:
            return None

        try:
            data = json.loads(message.decode("utf-8"))
        except ValueError:
            return None

        return data.get("password", encrypted)

    def ping(self):
        """
        Returns whether the authentication engine is reachable and working.
        """
        return self.state.ping()

    @property
    def federated_service(self):
        """
        Returns the name of the federated service for the auth system.

        If none, should return None.
        """
        return self.state.federated_service

    @property
    def requires_distinct_cli_password(self):
        """
        Returns whether this auth system requires a distinct CLI password to be created, in-system,
        before the CLI can be used.
        """
        return self.state.requires_distinct_cli_password

    @property
    def supports_encrypted_credentials(self):
        """
        Returns whether this auth system supports using encrypted credentials.
        """
        return self.state.supports_encrypted_credentials

    def has_password_set(self, username):
        """
        Returns whether the user has a password set in the auth system.
        """
        return self.state.has_password_set(username)

    @property
    def supports_fresh_login(self):
        """
        Returns whether this auth system supports the fresh login check.
        """
        return self.state.supports_fresh_login

    def query_users(self, query, limit=20):
        """
        Performs a lookup against the user system for the specified query. The returned tuple will
        be of the form (results, federated_login_id, err_msg). If the method is unsupported, the
        results portion of the tuple will be None instead of empty list.

        Note that this method can and will return results for users not yet found within the
        database; it is the responsibility of the caller to call link_user if they need the
        database row for the user system record.

        Results will be in the form of objects's with username and email fields.
        """
        return self.state.query_users(query, limit)

    def link_user(self, username_or_email):
        """
        Returns a tuple containing the database user record linked to the given username/email and
        any error that occurred when trying to link the user.
        """
        return self.state.link_user(username_or_email)

    def get_and_link_federated_user_info(self, user_info, internal_create=False):
        """
        Returns a tuple containing the database user record linked to the given UserInformation pair
        and any error that occurred when trying to link the user.

        If `internal_create` is True, the caller is an internal user creation process (such as team
        syncing), and the "can a user be created" check will be bypassed.
        """
        return self.state.get_and_link_federated_user_info(
            user_info, internal_create=internal_create
        )

    def confirm_existing_user(self, username, password):
        """
        Verifies that the given password matches to the given DB username.

        Unlike verify_credentials, this call first translates the DB user via the FederatedLogin
        table (where applicable).
        """
        return self.state.confirm_existing_user(username, password)

    def verify_credentials(self, username_or_email, password):
        """
        Verifies that the given username and password credentials are valid.
        """
        return self.state.verify_credentials(username_or_email, password)

    def check_group_lookup_args(self, group_lookup_args):
        """
        Verifies that the given group lookup args point to a valid group.

        Returns a tuple consisting of a boolean status and an error message (if any).
        """
        return self.state.check_group_lookup_args(group_lookup_args)

    def service_metadata(self):
        """
        Returns a dictionary of extra metadata to present to *superusers* about this auth engine.

        For example, LDAP returns the base DN so we can display to the user during sync setup.
        """
        return self.state.service_metadata()

    def iterate_group_members(self, group_lookup_args, page_size=None, disable_pagination=False):
        """
        Returns a tuple of an iterator over all the members of the group matching the given lookup
        args dictionary, or the error that occurred if the initial call failed or is unsupported.

        The format of the lookup args dictionary is specific to the implementation. Each result in
        the iterator is a tuple of (UserInformation, error_message), and only one will be not-None.
        """
        return self.state.iterate_group_members(
            group_lookup_args, page_size=page_size, disable_pagination=disable_pagination
        )

    def verify_and_link_user(self, username_or_email, password, basic_auth=False):
        """
        Verifies that the given username and password credentials are valid and, if so, creates or
        links the database user to the federated identity.
        """
        # First try to decode the password as a signed token.
        if basic_auth:
            decrypted = self._decrypt_user_password(password)
            if decrypted is None:
                # This is a normal password.
                if features.REQUIRE_ENCRYPTED_BASIC_AUTH:
                    msg = (
                        "Client login with unencrypted passwords is disabled. Please generate an "
                        + "encrypted password in the user admin panel for use here."
                    )
                    return (None, msg)
            else:
                password = decrypted

        (result, err_msg) = self.state.verify_and_link_user(username_or_email, password)
        if not result:
            return (result, err_msg)

        if not result.enabled:
            return (None, "This user has been disabled. Please contact your administrator.")

        return (result, err_msg)

    def is_superuser(self, username):
        return self.state.is_superuser(username)

    def is_global_readonly_superuser(self, username):
        return self.state.is_global_readonly_superuser(username)

    def has_superusers(self):
        return self.state.has_superusers()

    def is_restricted_user(self, username):
        return self.state.is_restricted_user(username)

    def has_restricted_users(self):
        return self.state.has_restricted_users()

    def __getattr__(self, name):
        return getattr(self.state, name, None)


class UserManager(object):
    def __init__(self, app, authentication: UserAuthentication):
        self.authentication = authentication
        self.state = self.init_app(app, authentication)

    def init_app(self, app, authentication: UserAuthentication):
        manager = None  # type: Union[FederatedUserManager, ConfigUserManager, None]
        if authentication.federated_service:
            manager = FederatedUserManager(app, authentication)
        else:
            manager = ConfigUserManager(app)

        return manager

    def __getattr__(self, name):
        return getattr(self.state, name, None)

    def is_restricted_user(self, username):
        if not features.RESTRICTED_USERS:
            return False

        return self.state.is_restricted_user(username)

    def is_superuser(self, username):
        if not features.SUPER_USERS:
            return False

        return self.state.is_superuser(username)


class FederatedUserManager(ConfigUserManager):
    """
    Helper class to access superusers defined in the backed federated provider.
    Falls back to the in-memory helper if not available in the federated service.
    """

    def __init__(self, app, authentication: UserAuthentication):
        self.federated_users = authentication
        super().__init__(app)

    def __get_federated_login_identifier(self, username) -> str:
        db_user = model.user.get_user(username)
        if not db_user:
            return ""

        federated_login = model.user.lookup_federated_login(
            db_user, self.federated_users.federated_service
        )
        if not federated_login:
            return ""
        return federated_login.service_ident

    def is_superuser(self, username: str) -> bool:
        """
        Returns if the given username represents a super user.
        """
        identifier = self.__get_federated_login_identifier(username)
        if not identifier:
            identifier = username

        return self.federated_users.is_superuser(identifier) or super().is_superuser(username)

    def has_superusers(self) -> bool:
        """
        Returns whether there are any superusers defined.
        """
        return self.federated_users.has_superusers() or super().has_superusers()

    def is_restricted_user(self, username: str, include_robots: bool = True) -> bool:
        """
        Returns if the given username represents a restricted user.
        """
        if include_robots:
            username = username.split("+", 1)[0]

        if super().restricted_whitelist_is_set() and not super().is_restricted_user(username):
            return False

        identifier = self.__get_federated_login_identifier(username)
        if not identifier:
            identifier = username

        return self.federated_users.is_restricted_user(identifier) or super().is_restricted_user(
            username
        )

    def has_restricted_users(self) -> bool:
        return self.federated_users.has_restricted_users() or super().has_restricted_users()

    def is_global_readonly_superuser(self, username: str) -> bool:
        identifier = self.__get_federated_login_identifier(username)
        if not identifier:
            identifier = username

        return self.federated_users.is_global_readonly_superuser(
            identifier
        ) or super().is_global_readonly_superuser(username)