powerdns-admin/powerdnsadmin/models/user.py

import os
import base64
import traceback
import bcrypt
import pyotp
import ldap
import ldap.filter
from flask import current_app
from flask_login import AnonymousUserMixin
from sqlalchemy import orm

from .base import db
from .role import Role
from .setting import Setting
from .domain_user import DomainUser


class Anonymous(AnonymousUserMixin):
    def __init__(self):
        self.username = 'Anonymous'


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), index=True, unique=True)
    password = db.Column(db.String(64))
    firstname = db.Column(db.String(64))
    lastname = db.Column(db.String(64))
    email = db.Column(db.String(128))
    otp_secret = db.Column(db.String(16))
    confirmed = db.Column(db.SmallInteger, nullable=False, default=0)
    role_id = db.Column(db.Integer, db.ForeignKey('role.id'))
    accounts = None

    def __init__(self,
                 id=None,
                 username=None,
                 password=None,
                 plain_text_password=None,
                 firstname=None,
                 lastname=None,
                 role_id=None,
                 email=None,
                 otp_secret=None,
                 confirmed=False,
                 reload_info=True):
        self.id = id
        self.username = username
        self.password = password
        self.plain_text_password = plain_text_password
        self.firstname = firstname
        self.lastname = lastname
        self.role_id = role_id
        self.email = email
        self.otp_secret = otp_secret
        self.confirmed = confirmed

        if reload_info:
            user_info = self.get_user_info_by_id(
            ) if id else self.get_user_info_by_username()

            if user_info:
                self.id = user_info.id
                self.username = user_info.username
                self.firstname = user_info.firstname
                self.lastname = user_info.lastname
                self.email = user_info.email
                self.role_id = user_info.role_id
                self.otp_secret = user_info.otp_secret
                self.confirmed = user_info.confirmed

    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

    def get_id(self):
        try:
            return unicode(self.id)  # python 2
        except NameError:
            return str(self.id)  # python 3

    def __repr__(self):
        return '<User {0}>'.format(self.username)

    def get_totp_uri(self):
        return "otpauth://totp/PowerDNS-Admin:{0}?secret={1}&issuer=PowerDNS-Admin".format(
            self.username, self.otp_secret)

    def verify_totp(self, token):
        totp = pyotp.TOTP(self.otp_secret)
        return totp.verify(token)

    def get_hashed_password(self, plain_text_password=None):
        # Hash a password for the first time
        #   (Using bcrypt, the salt is saved into the hash itself)
        if plain_text_password is None:
            return plain_text_password

        pw = plain_text_password if plain_text_password else self.plain_text_password
        return bcrypt.hashpw(pw.encode('utf-8'), bcrypt.gensalt())

    def check_password(self, hashed_password):
        # Check hashed password. Using bcrypt, the salt is saved into the hash itself
        if (self.plain_text_password):
            return bcrypt.checkpw(self.plain_text_password.encode('utf-8'),
                                  hashed_password.encode('utf-8'))
        return False

    def get_user_info_by_id(self):
        user_info = User.query.get(int(self.id))
        return user_info

    def get_user_info_by_username(self):
        user_info = User.query.filter(User.username == self.username).first()
        return user_info

    def ldap_init_conn(self):
        ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
        conn = ldap.initialize(Setting().get('ldap_uri'))
        conn.set_option(ldap.OPT_REFERRALS, ldap.OPT_OFF)
        conn.set_option(ldap.OPT_PROTOCOL_VERSION, 3)
        conn.set_option(ldap.OPT_X_TLS, ldap.OPT_X_TLS_DEMAND)
        conn.set_option(ldap.OPT_X_TLS_DEMAND, True)
        conn.set_option(ldap.OPT_DEBUG_LEVEL, 255)
        conn.protocol_version = ldap.VERSION3
        return conn

    def ldap_search(self, searchFilter, baseDN):
        searchScope = ldap.SCOPE_SUBTREE
        retrieveAttributes = None

        try:
            conn = self.ldap_init_conn()
            if Setting().get('ldap_type') == 'ad':
                conn.simple_bind_s(
                    "{0}@{1}".format(self.username,
                                     Setting().get('ldap_domain')),
                    self.password)
            else:
                conn.simple_bind_s(Setting().get('ldap_admin_username'),
                                   Setting().get('ldap_admin_password'))
            ldap_result_id = conn.search(baseDN, searchScope, searchFilter,
                                         retrieveAttributes)
            result_set = []

            while 1:
                result_type, result_data = conn.result(ldap_result_id, 0)
                if (result_data == []):
                    break
                else:
                    if result_type == ldap.RES_SEARCH_ENTRY:
                        result_set.append(result_data)
            return result_set

        except ldap.LDAPError as e:
            current_app.logger.error(e)
            current_app.logger.debug('baseDN: {0}'.format(baseDN))
            current_app.logger.debug(traceback.format_exc())

    def ldap_auth(self, ldap_username, password):
        try:
            conn = self.ldap_init_conn()
            conn.simple_bind_s(ldap_username, password)
            return True
        except ldap.LDAPError as e:
            current_app.logger.error(e)
            return False

    def ad_recursive_groups(self, groupDN):
        """
        Recursively list groups belonging to a group. It will allow checking deep in the Active Directory
        whether a user is allowed to enter or not
        """
        LDAP_BASE_DN = Setting().get('ldap_base_dn')
        groupSearchFilter = "(&(objectcategory=group)(member=%s))" % ldap.filter.escape_filter_chars(
            groupDN)
        result = [groupDN]
        try:
            groups = self.ldap_search(groupSearchFilter, LDAP_BASE_DN)
            for group in groups:
                result += [group[0][0]]
                if 'memberOf' in group[0][1]:
                    for member in group[0][1]['memberOf']:
                        result += self.ad_recursive_groups(
                            member.decode("utf-8"))
            return result
        except ldap.LDAPError as e:
            current_app.logger.exception("Recursive AD Group search error")
            return result

    def is_validate(self, method, src_ip='', trust_user=False):
        """
        Validate user credential
        """
        role_name = 'User'

        if method == 'LOCAL':
            user_info = User.query.filter(
                User.username == self.username).first()

            if user_info:
                if trust_user or (user_info.password and self.check_password(
                        user_info.password)):
                    current_app.logger.info(
                        'User "{0}" logged in successfully. Authentication request from {1}'
                        .format(self.username, src_ip))
                    return True
                current_app.logger.error(
                    'User "{0}" inputted a wrong password. Authentication request from {1}'
                    .format(self.username, src_ip))
                return False

            current_app.logger.warning(
                'User "{0}" does not exist. Authentication request from {1}'.
                format(self.username, src_ip))
            return False

        if method == 'LDAP':
            LDAP_TYPE = Setting().get('ldap_type')
            LDAP_BASE_DN = Setting().get('ldap_base_dn')
            LDAP_FILTER_BASIC = Setting().get('ldap_filter_basic')
            LDAP_FILTER_USERNAME = Setting().get('ldap_filter_username')
            LDAP_FILTER_GROUP = Setting().get('ldap_filter_group')
            LDAP_FILTER_GROUPNAME = Setting().get('ldap_filter_groupname')
            LDAP_ADMIN_GROUP = Setting().get('ldap_admin_group')
            LDAP_OPERATOR_GROUP = Setting().get('ldap_operator_group')
            LDAP_USER_GROUP = Setting().get('ldap_user_group')
            LDAP_GROUP_SECURITY_ENABLED = Setting().get('ldap_sg_enabled')

            # validate AD user password
            if Setting().get('ldap_type') == 'ad' and not trust_user:
                ldap_username = "{0}@{1}".format(self.username,
                                                 Setting().get('ldap_domain'))
                if not self.ldap_auth(ldap_username, self.password):
                    current_app.logger.error(
                        'User "{0}" input a wrong LDAP password. Authentication request from {1}'
                        .format(self.username, src_ip))
                    return False

            searchFilter = "(&({0}={1}){2})".format(LDAP_FILTER_USERNAME,
                                                    self.username,
                                                    LDAP_FILTER_BASIC)
            current_app.logger.debug('Ldap searchFilter {0}'.format(searchFilter))

            ldap_result = self.ldap_search(searchFilter, LDAP_BASE_DN)
            current_app.logger.debug('Ldap search result: {0}'.format(ldap_result))

            if not ldap_result:
                current_app.logger.warning(
                    'LDAP User "{0}" does not exist. Authentication request from {1}'
                    .format(self.username, src_ip))
                return False
            else:
                try:
                    ldap_username = ldap.filter.escape_filter_chars(
                        ldap_result[0][0][0])

                    if Setting().get('ldap_type') != 'ad' and not trust_user:
                        # validate ldap user password
                        if not self.ldap_auth(ldap_username, self.password):
                            current_app.logger.error(
                                'User "{0}" input a wrong LDAP password. Authentication request from {1}'
                                .format(self.username, src_ip))
                            return False

                    # check if LDAP_GROUP_SECURITY_ENABLED is True
                    # user can be assigned to ADMIN or USER role.
                    if LDAP_GROUP_SECURITY_ENABLED:
                        try:
                            if LDAP_TYPE == 'ldap':
                                groupSearchFilter = "(&({0}={1}){2})".format(LDAP_FILTER_GROUPNAME, ldap_username, LDAP_FILTER_GROUP)
                                current_app.logger.debug('Ldap groupSearchFilter {0}'.format(groupSearchFilter))
                                if (self.ldap_search(groupSearchFilter,
                                                     LDAP_ADMIN_GROUP)):
                                    role_name = 'Administrator'
                                    current_app.logger.info(
                                        'User {0} is part of the "{1}" group that allows admin access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_ADMIN_GROUP))
                                elif (self.ldap_search(groupSearchFilter,
                                                       LDAP_OPERATOR_GROUP)):
                                    role_name = 'Operator'
                                    current_app.logger.info(
                                        'User {0} is part of the "{1}" group that allows operator access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_OPERATOR_GROUP))
                                elif (self.ldap_search(groupSearchFilter,
                                                       LDAP_USER_GROUP)):
                                    current_app.logger.info(
                                        'User {0} is part of the "{1}" group that allows user access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_USER_GROUP))
                                else:
                                    current_app.logger.error(
                                        'User {0} is not part of the "{1}", "{2}" or "{3}" groups that allow access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_ADMIN_GROUP,
                                                LDAP_OPERATOR_GROUP,
                                                LDAP_USER_GROUP))
                                    return False
                            elif LDAP_TYPE == 'ad':
                                user_ldap_groups = []
                                user_ad_member_of = ldap_result[0][0][1].get(
                                    'memberOf')

                                if not user_ad_member_of:
                                    current_app.logger.error(
                                        'User {0} does not belong to any group while LDAP_GROUP_SECURITY_ENABLED is ON'
                                        .format(self.username))
                                    return False

                                for group in [
                                        g.decode("utf-8")
                                        for g in user_ad_member_of
                                ]:
                                    user_ldap_groups += self.ad_recursive_groups(
                                        group)

                                if (LDAP_ADMIN_GROUP in user_ldap_groups):
                                    role_name = 'Administrator'
                                    current_app.logger.info(
                                        'User {0} is part of the "{1}" group that allows admin access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_ADMIN_GROUP))
                                elif (LDAP_OPERATOR_GROUP in user_ldap_groups):
                                    role_name = 'Operator'
                                    current_app.logger.info(
                                        'User {0} is part of the "{1}" group that allows operator access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_OPERATOR_GROUP))
                                elif (LDAP_USER_GROUP in user_ldap_groups):
                                    current_app.logger.info(
                                        'User {0} is part of the "{1}" group that allows user access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_USER_GROUP))
                                else:
                                    current_app.logger.error(
                                        'User {0} is not part of the "{1}", "{2}" or "{3}" groups that allow access to PowerDNS-Admin'
                                        .format(self.username,
                                                LDAP_ADMIN_GROUP,
                                                LDAP_OPERATOR_GROUP,
                                                LDAP_USER_GROUP))
                                    return False
                            else:
                                current_app.logger.error('Invalid LDAP type')
                                return False
                        except Exception as e:
                            current_app.logger.error(
                                'LDAP group lookup for user "{0}" has failed. Authentication request from {1}'
                                .format(self.username, src_ip))
                            current_app.logger.debug(traceback.format_exc())
                            return False

                except Exception as e:
                    current_app.logger.error('Wrong LDAP configuration. {0}'.format(e))
                    current_app.logger.debug(traceback.format_exc())
                    return False

            # create user if not exist in the db
            if not User.query.filter(User.username == self.username).first():
                self.firstname = self.username
                self.lastname = ''
                try:
                    # try to get user's firstname, lastname and email address from LDAP attributes
                    if LDAP_TYPE == 'ldap':
                        self.firstname = ldap_result[0][0][1]['givenName'][
                            0].decode("utf-8")
                        self.lastname = ldap_result[0][0][1]['sn'][0].decode(
                            "utf-8")
                        self.email = ldap_result[0][0][1]['mail'][0].decode(
                            "utf-8")
                    elif LDAP_TYPE == 'ad':
                        self.firstname = ldap_result[0][0][1]['name'][
                            0].decode("utf-8")
                        self.email = ldap_result[0][0][1]['userPrincipalName'][
                            0].decode("utf-8")
                except Exception as e:
                    current_app.logger.warning(
                        "Reading ldap data threw an exception {0}".format(e))
                    current_app.logger.debug(traceback.format_exc())

                # first register user will be in Administrator role
                if User.query.count() == 0:
                    self.role_id = Role.query.filter_by(
                        name='Administrator').first().id
                else:
                    self.role_id = Role.query.filter_by(
                        name=role_name).first().id

                self.create_user()
                current_app.logger.info('Created user "{0}" in the DB'.format(
                    self.username))

            # user already exists in database, set their role based on group membership (if enabled)
            if LDAP_GROUP_SECURITY_ENABLED:
                self.set_role(role_name)

            return True
        else:
            current_app.logger.error('Unsupported authentication method')
            return False

    def create_user(self):
        """
        If user logged in successfully via LDAP in the first time
        We will create a local user (in DB) in order to manage user
        profile such as name, roles,...
        """

        # Set an invalid password hash for non local users
        self.password = '*'

        db.session.add(self)
        db.session.commit()

    def create_local_user(self):
        """
        Create local user witch stores username / password in the DB
        """
        # check if username existed
        user = User.query.filter(User.username == self.username).first()
        if user:
            return {'status': False, 'msg': 'Username is already in use'}

        # check if email existed
        user = User.query.filter(User.email == self.email).first()
        if user:
            return {'status': False, 'msg': 'Email address is already in use'}

        # first register user will be in Administrator role
        if self.role_id is None:
            self.role_id = Role.query.filter_by(name='User').first().id
        if User.query.count() == 0:
            self.role_id = Role.query.filter_by(
                name='Administrator').first().id

        self.password = self.get_hashed_password(
            self.plain_text_password) if self.plain_text_password else '*'

        if self.password and self.password != '*':
            self.password = self.password.decode("utf-8")

        db.session.add(self)
        db.session.commit()
        return {'status': True, 'msg': 'Created user successfully'}

    def update_local_user(self):
        """
        Update local user
        """
        # Sanity check - account name
        if self.username == "":
            return {'status': False, 'msg': 'No user name specified'}

        # read user and check that it exists
        user = User.query.filter(User.username == self.username).first()
        if not user:
            return {'status': False, 'msg': 'User does not exist'}

        # check if new email exists (only if changed)
        if user.email != self.email:
            checkuser = User.query.filter(User.email == self.email).first()
            if checkuser:
                return {
                    'status': False,
                    'msg': 'New email address is already in use'
                }

        user.firstname = self.firstname
        user.lastname = self.lastname
        user.email = self.email

        # store new password hash (only if changed)
        if self.plain_text_password:
            user.password = self.get_hashed_password(
                self.plain_text_password).decode("utf-8")

        db.session.commit()
        return {'status': True, 'msg': 'User updated successfully'}

    def update_profile(self, enable_otp=None):
        """
        Update user profile
        """

        user = User.query.filter(User.username == self.username).first()
        if not user:
            return False

        user.firstname = self.firstname if self.firstname else user.firstname
        user.lastname = self.lastname if self.lastname else user.lastname
        user.password = self.get_hashed_password(
            self.plain_text_password).decode(
                "utf-8") if self.plain_text_password else user.password

        if self.email:
            # Can not update to a new email that
            # already been used.
            existing_email = User.query.filter(
                User.email == self.email,
                User.username != self.username).first()
            if existing_email:
                return False
            # If need to verify new email,
            # update the "confirmed" status.
            if user.email != self.email:
                user.email = self.email
                if Setting().get('verify_user_email'):
                    user.confirmed = 0

        if enable_otp is not None:
            user.otp_secret = ""

        if enable_otp == True:
            # generate the opt secret key
            user.otp_secret = base64.b32encode(os.urandom(10)).decode('utf-8')

        try:
            db.session.add(user)
            db.session.commit()
            return True
        except Exception:
            db.session.rollback()
            return False

    def update_confirmed(self, confirmed):
        """
        Update user email confirmation status
        """
        self.confirmed = confirmed
        db.session.commit()

    def get_domains(self):
        """
        Get list of domains which the user is granted to have
        access.

        Note: This doesn't include the permission granting from Account
        which user belong to
        """

        return self.get_domain_query().all()

    def delete(self):
        """
        Delete a user
        """
        # revoke all user privileges first
        self.revoke_privilege()

        try:
            User.query.filter(User.username == self.username).delete()
            db.session.commit()
            return True
        except Exception as e:
            db.session.rollback()
            current_app.logger.error('Cannot delete user {0} from DB. DETAIL: {1}'.format(
                self.username, e))
            return False

    def revoke_privilege(self):
        """
        Revoke all privileges from a user
        """
        user = User.query.filter(User.username == self.username).first()

        if user:
            user_id = user.id
            try:
                DomainUser.query.filter(DomainUser.user_id == user_id).delete()
                db.session.commit()
                return True
            except Exception as e:
                db.session.rollback()
                current_app.logger.error(
                    'Cannot revoke user {0} privileges. DETAIL: {1}'.format(
                        self.username, e))
                return False
        return False

    def set_role(self, role_name):
        role = Role.query.filter(Role.name == role_name).first()
        if role:
            user = User.query.filter(User.username == self.username).first()
            user.role_id = role.id
            db.session.commit()
            return {'status': True, 'msg': 'Set user role successfully'}
        else:
            return {'status': False, 'msg': 'Role does not exist'}

    @orm.reconstructor
    def set_account(self):
        self.accounts = self.get_accounts()

    def get_accounts(self):
        """
        Get accounts associated with this user
        """
        from .account import Account
        from .account_user import AccountUser
        accounts = []
        query = db.session\
            .query(
                AccountUser,
                Account)\
            .filter(self.id == AccountUser.user_id)\
            .filter(Account.id == AccountUser.account_id)\
            .all()
        for q in query:
            accounts.append(q[1])
        return accounts