quay/data/model/user.py

import bcrypt
import logging
import json

from peewee import JOIN_LEFT_OUTER, IntegrityError, fn
from uuid import uuid4
from datetime import datetime, timedelta

from data.database import (User, LoginService, FederatedLogin, RepositoryPermission, TeamMember,
                           Team, Repository, TupleSelector, TeamRole, Namespace, Visibility,
                           EmailConfirmation, Role, db_for_update, random_string_generator,
                           UserRegion, ImageStorageLocation)
from data.model import (DataModelException, InvalidPasswordException, InvalidRobotException,
                        InvalidUsernameException, InvalidEmailAddressException,
                        TooManyUsersException, TooManyLoginAttemptsException, db_transaction,
                        notification, config, repository, _basequery)
from util.names import format_robot_username, parse_robot_username
from util.validation import (validate_username, validate_email, validate_password,
                             INVALID_PASSWORD_MESSAGE)
from util.backoff import exponential_backoff


logger = logging.getLogger(__name__)


EXPONENTIAL_BACKOFF_SCALE = timedelta(seconds=1)


def hash_password(password, salt=None):
  salt = salt or bcrypt.gensalt()
  return bcrypt.hashpw(password.encode('utf-8'), salt)


def is_create_user_allowed():
  return True


def create_user(username, password, email, auto_verify=False):
  """ Creates a regular user, if allowed. """
  if not validate_password(password):
    raise InvalidPasswordException(INVALID_PASSWORD_MESSAGE)

  if not is_create_user_allowed():
    raise TooManyUsersException()

  created = create_user_noverify(username, email)
  created.password_hash = hash_password(password)
  created.verified = auto_verify
  created.save()

  return created


def create_user_noverify(username, email):
  if not validate_email(email):
    raise InvalidEmailAddressException('Invalid email address: %s' % email)

  (username_valid, username_issue) = validate_username(username)
  if not username_valid:
    raise InvalidUsernameException('Invalid username %s: %s' % (username, username_issue))

  try:
    existing = User.get((User.username == username) | (User.email == email))

    logger.info('Existing user with same username or email.')

    # A user already exists with either the same username or email
    if existing.username == username:
      raise InvalidUsernameException('Username has already been taken: %s' %
                                     username)
    raise InvalidEmailAddressException('Email has already been used: %s' %
                                       email)

  except User.DoesNotExist:
    # This is actually the happy path
    logger.debug('Email and username are unique!')

  try:
    return User.create(username=username, email=email)
  except Exception as ex:
    raise DataModelException(ex.message)


def is_username_unique(test_username):
  try:
    User.get((User.username == test_username))
    return False
  except User.DoesNotExist:
    return True


def change_password(user, new_password):
  if not validate_password(new_password):
    raise InvalidPasswordException(INVALID_PASSWORD_MESSAGE)

  pw_hash = hash_password(new_password)
  user.invalid_login_attempts = 0
  user.password_hash = pw_hash
  user.uuid = str(uuid4())
  user.save()

  # Remove any password required notifications for the user.
  notification.delete_notifications_by_kind(user, 'password_required')


def change_username(user_id, new_username):
  (username_valid, username_issue) = validate_username(new_username)
  if not username_valid:
    raise InvalidUsernameException('Invalid username %s: %s' % (new_username, username_issue))

  with db_transaction():
    # Reload the user for update
    user = db_for_update(User.select().where(User.id == user_id)).get()

    # Rename the robots
    for robot in db_for_update(_list_entity_robots(user.username)):
      _, robot_shortname = parse_robot_username(robot.username)
      new_robot_name = format_robot_username(new_username, robot_shortname)
      robot.username = new_robot_name
      robot.save()

    # Rename the user
    user.username = new_username
    user.save()
    return user


def change_invoice_email_address(user, invoice_email_address):
  # Note: We null out the address if it is an empty string.
  user.invoice_email_address = invoice_email_address or None
  user.save()


def change_send_invoice_email(user, invoice_email):
  user.invoice_email = invoice_email
  user.save()


def change_user_tag_expiration(user, tag_expiration_s):
  user.removed_tag_expiration_s = tag_expiration_s
  user.save()


def update_email(user, new_email, auto_verify=False):
  try:
    user.email = new_email
    user.verified = auto_verify
    user.save()
  except IntegrityError:
    raise DataModelException('E-mail address already used')


def create_robot(robot_shortname, parent):
  (username_valid, username_issue) = validate_username(robot_shortname)
  if not username_valid:
    raise InvalidRobotException('The name for the robot \'%s\' is invalid: %s' %
                                (robot_shortname, username_issue))

  username = format_robot_username(parent.username, robot_shortname)

  try:
    User.get(User.username == username)

    msg = 'Existing robot with name: %s' % username
    logger.info(msg)
    raise InvalidRobotException(msg)

  except User.DoesNotExist:
    pass

  try:
    created = User.create(username=username, robot=True)

    service = LoginService.get(name='quayrobot')
    password = created.email
    FederatedLogin.create(user=created, service=service,
                          service_ident=password)

    return created, password
  except Exception as ex:
    raise DataModelException(ex.message)


def get_robot(robot_shortname, parent):
  robot_username = format_robot_username(parent.username, robot_shortname)
  robot = lookup_robot(robot_username)
  return robot, robot.email


def lookup_robot(robot_username):
  try:
    return (User
            .select()
            .join(FederatedLogin)
            .join(LoginService)
            .where(LoginService.name == 'quayrobot', User.username == robot_username,
                   User.robot == True)
            .get())
  except User.DoesNotExist:
    raise InvalidRobotException('Could not find robot with username: %s' % robot_username)


def get_matching_robots(name_prefix, username, limit=10):
  admined_orgs = (_basequery.get_user_organizations(username)
                  .switch(Team)
                  .join(TeamRole)
                  .where(TeamRole.name == 'admin'))

  prefix_checks = False

  for org in admined_orgs:
    org_search =  _basequery.prefix_search(User.username, org.username + '+' + name_prefix)
    prefix_checks = prefix_checks | org_search

  user_search =  _basequery.prefix_search(User.username, username + '+' + name_prefix)
  prefix_checks = prefix_checks | user_search

  return User.select().where(prefix_checks).limit(limit)


def verify_robot(robot_username, password):
  result = parse_robot_username(robot_username)
  if result is None:
    raise InvalidRobotException('%s is an invalid robot name' % robot_username)

  # Find the matching robot.
  query = (User
           .select()
           .join(FederatedLogin)
           .join(LoginService)
           .where(FederatedLogin.service_ident == password, LoginService.name == 'quayrobot',
                  User.username == robot_username))

  try:
    robot = query.get()
  except User.DoesNotExist:
    msg = ('Could not find robot with username: %s and supplied password.' %
           robot_username)
    raise InvalidRobotException(msg)

  # Find the owner user and ensure it is not disabled.
  try:
    owner = User.get(User.username == result[0])
  except User.DoesNotExist:
    raise InvalidRobotException('Robot %s owner does not exist' % robot_username)

  if not owner.enabled:
    raise InvalidRobotException('This user has been disabled. Please contact your administrator.')

  return robot

def regenerate_robot_token(robot_shortname, parent):
  robot_username = format_robot_username(parent.username, robot_shortname)

  robot = lookup_robot(robot_username)
  password = random_string_generator(length=64)()
  robot.email = password
  robot.uuid = str(uuid4())

  service = LoginService.get(name='quayrobot')
  login = FederatedLogin.get(FederatedLogin.user == robot, FederatedLogin.service == service)
  login.service_ident = password

  login.save()
  robot.save()

  return robot, password

def delete_robot(robot_username):
  try:
    robot = User.get(username=robot_username, robot=True)
    robot.delete_instance(recursive=True, delete_nullable=True)

  except User.DoesNotExist:
    raise InvalidRobotException('Could not find robot with username: %s' %
                                robot_username)


def _list_entity_robots(entity_name):
  """ Return the list of robots for the specified entity. This MUST return a query, not a
      materialized list so that callers can use db_for_update.
      """
  return (User
          .select()
          .join(FederatedLogin)
          .where(User.robot == True, User.username ** (entity_name + '+%')))


def list_entity_robot_permission_teams(entity_name, include_permissions=False):
  query = (_list_entity_robots(entity_name))

  fields = [User.username, FederatedLogin.service_ident]
  if include_permissions:
    query = (query
             .join(RepositoryPermission, JOIN_LEFT_OUTER,
                   on=(RepositoryPermission.user == FederatedLogin.user))
             .join(Repository, JOIN_LEFT_OUTER)
             .switch(User)
             .join(TeamMember, JOIN_LEFT_OUTER)
             .join(Team, JOIN_LEFT_OUTER))

    fields.append(Repository.name)
    fields.append(Team.name)

  return TupleSelector(query, fields)


def confirm_attached_federated_login(user, service_name):
  """ Verifies that the given user has a federated service identity for the specified service.
      If none found, a row is added for that service and user.
  """
  with db_transaction():
    if not lookup_federated_login(user, service_name):
      attach_federated_login(user, service_name, user.username)


def create_federated_user(username, email, service_name, service_id,
                          set_password_notification, metadata={}):
  if not is_create_user_allowed():
    raise TooManyUsersException()

  new_user = create_user_noverify(username, email)
  new_user.verified = True
  new_user.save()

  service = LoginService.get(LoginService.name == service_name)
  FederatedLogin.create(user=new_user, service=service,
                        service_ident=service_id,
                        metadata_json=json.dumps(metadata))

  if set_password_notification:
    notification.create_notification('password_required', new_user)

  return new_user


def attach_federated_login(user, service_name, service_id, metadata={}):
  service = LoginService.get(LoginService.name == service_name)
  FederatedLogin.create(user=user, service=service, service_ident=service_id,
                        metadata_json=json.dumps(metadata))
  return user


def verify_federated_login(service_name, service_id):
  try:
    found = (FederatedLogin
             .select(FederatedLogin, User)
             .join(LoginService)
             .switch(FederatedLogin).join(User)
             .where(FederatedLogin.service_ident == service_id, LoginService.name == service_name)
             .get())
    return found.user
  except FederatedLogin.DoesNotExist:
    return None


def list_federated_logins(user):
  selected = FederatedLogin.select(FederatedLogin.service_ident,
                                   LoginService.name, FederatedLogin.metadata_json)
  joined = selected.join(LoginService)
  return joined.where(LoginService.name != 'quayrobot',
                      FederatedLogin.user == user)


def lookup_federated_login(user, service_name):
  try:
    return list_federated_logins(user).where(LoginService.name == service_name).get()
  except FederatedLogin.DoesNotExist:
    return None

def create_confirm_email_code(user, new_email=None):
  if new_email:
    if not validate_email(new_email):
      raise InvalidEmailAddressException('Invalid email address: %s' %
                                         new_email)

  code = EmailConfirmation.create(user=user, email_confirm=True,
                                  new_email=new_email)
  return code


def confirm_user_email(code):
  try:
    code = EmailConfirmation.get(EmailConfirmation.code == code,
                                 EmailConfirmation.email_confirm == True)
  except EmailConfirmation.DoesNotExist:
    raise DataModelException('Invalid email confirmation code.')

  user = code.user
  user.verified = True

  old_email = None
  new_email = code.new_email
  if new_email and new_email != old_email:
    if find_user_by_email(new_email):
      raise DataModelException('E-mail address already used.')

    old_email = user.email
    user.email = new_email

  user.save()

  code.delete_instance()

  return user, new_email, old_email


def create_reset_password_email_code(email):
  try:
    user = User.get(User.email == email)
  except User.DoesNotExist:
    raise InvalidEmailAddressException('Email address was not found.')

  if user.organization:
    raise InvalidEmailAddressException('Organizations can not have passwords.')

  code = EmailConfirmation.create(user=user, pw_reset=True)
  return code


def validate_reset_code(code):
  try:
    code = EmailConfirmation.get(EmailConfirmation.code == code,
                                 EmailConfirmation.pw_reset == True)
  except EmailConfirmation.DoesNotExist:
    return None

  user = code.user
  code.delete_instance()

  return user


def find_user_by_email(email):
  try:
    return User.get(User.email == email)
  except User.DoesNotExist:
    return None


def get_nonrobot_user(username):
  try:
    return User.get(User.username == username, User.organization == False, User.robot == False)
  except User.DoesNotExist:
    return None


def get_user(username):
  try:
    return User.get(User.username == username, User.organization == False)
  except User.DoesNotExist:
    return None


def get_namespace_user(username):
  try:
    return User.get(User.username == username)
  except User.DoesNotExist:
    return None


def get_user_or_org(username):
  try:
    return User.get(User.username == username, User.robot == False)
  except User.DoesNotExist:
    return None


def get_user_by_id(user_db_id):
  try:
    return User.get(User.id == user_db_id, User.organization == False)
  except User.DoesNotExist:
    return None


def get_namespace_user_by_user_id(namespace_user_db_id):
  try:
    return User.get(User.id == namespace_user_db_id, User.robot == False)
  except User.DoesNotExist:
    raise InvalidUsernameException('User with id does not exist: %s' % namespace_user_db_id)


def get_namespace_by_user_id(namespace_user_db_id):
  try:
    return User.get(User.id == namespace_user_db_id, User.robot == False).username
  except User.DoesNotExist:
    raise InvalidUsernameException('User with id does not exist: %s' % namespace_user_db_id)


def get_user_by_uuid(user_uuid):
  try:
    return User.get(User.uuid == user_uuid, User.organization == False)
  except User.DoesNotExist:
    return None


def get_user_or_org_by_customer_id(customer_id):
  try:
    return User.get(User.stripe_id == customer_id)
  except User.DoesNotExist:
    return None


def get_matching_user_namespaces(namespace_prefix, username, limit=10):
  namespace_search = _basequery.prefix_search(Namespace.username, namespace_prefix)
  base_query = (Namespace
                .select()
                .distinct()
                .join(Repository, on=(Repository.namespace_user == Namespace.id))
                .join(RepositoryPermission, JOIN_LEFT_OUTER)
                .where(namespace_search))

  return _basequery.filter_to_repos_for_user(base_query, username).limit(limit)

def get_matching_users(username_prefix, robot_namespace=None,
                       organization=None):
  user_search =  _basequery.prefix_search(User.username, username_prefix)
  direct_user_query = (user_search & (User.organization == False) & (User.robot == False))

  if robot_namespace:
    robot_prefix = format_robot_username(robot_namespace, username_prefix)
    robot_search = _basequery.prefix_search(User.username, robot_prefix)
    direct_user_query = (direct_user_query | (robot_search & (User.robot == True)))

  query = (User
           .select(User.username, User.email, User.robot)
           .group_by(User.username, User.email, User.robot)
           .where(direct_user_query))

  if organization:
    query = (query
             .select(User.username, User.email, User.robot, fn.Sum(Team.id))
             .join(TeamMember, JOIN_LEFT_OUTER)
             .join(Team, JOIN_LEFT_OUTER, on=((Team.id == TeamMember.team) &
                                              (Team.organization == organization))))

  class MatchingUserResult(object):
    def __init__(self, *args):
      self.username = args[0]
      self.email = args[1]
      self.robot = args[2]

      if organization:
        self.is_org_member = (args[3] != None)
      else:
        self.is_org_member = None

  return (MatchingUserResult(*args) for args in query.tuples().limit(10))


def verify_user(username_or_email, password):
  # Make sure we didn't get any unicode for the username.
  try:
    str(username_or_email)
  except ValueError:
    return None

  try:
    fetched = User.get((User.username == username_or_email) |
                       (User.email == username_or_email))
  except User.DoesNotExist:
    return None

  now = datetime.utcnow()

  if fetched.invalid_login_attempts > 0:
    can_retry_at = exponential_backoff(fetched.invalid_login_attempts, EXPONENTIAL_BACKOFF_SCALE,
                                       fetched.last_invalid_login)

    if can_retry_at > now:
      retry_after = can_retry_at - now
      raise TooManyLoginAttemptsException('Too many login attempts.', retry_after.total_seconds())

  if (fetched.password_hash and
      hash_password(password, fetched.password_hash) == fetched.password_hash):
    if fetched.invalid_login_attempts > 0:
      fetched.invalid_login_attempts = 0
      fetched.save()

    return fetched

  fetched.invalid_login_attempts += 1
  fetched.last_invalid_login = now
  fetched.save()

  # We weren't able to authorize the user
  return None


def get_all_repo_users(namespace_name, repository_name):
  return (RepositoryPermission
          .select(User.username, User.email, User.robot, Role.name, RepositoryPermission)
          .join(User)
          .switch(RepositoryPermission)
          .join(Role)
          .switch(RepositoryPermission)
          .join(Repository)
          .join(Namespace, on=(Repository.namespace_user == Namespace.id))
          .where(Namespace.username == namespace_name, Repository.name == repository_name))


def get_all_repo_users_transitive_via_teams(namespace_name, repository_name):
  return (User
          .select()
          .distinct()
          .join(TeamMember)
          .join(Team)
          .join(RepositoryPermission)
          .join(Repository)
          .join(Namespace, on=(Repository.namespace_user == Namespace.id))
          .where(Namespace.username == namespace_name, Repository.name == repository_name))


def get_all_repo_users_transitive(namespace_name, repository_name):
  # Load the users found via teams and directly via permissions.
  via_teams = get_all_repo_users_transitive_via_teams(namespace_name, repository_name)
  directly = [perm.user for perm in get_all_repo_users(namespace_name, repository_name)]

  # Filter duplicates.
  user_set = set()

  def check_add(u):
    if u.username in user_set:
      return False

    user_set.add(u.username)
    return True

  return [user for user in list(directly) + list(via_teams) if check_add(user)]


def get_private_repo_count(username):
  return (Repository
          .select()
          .join(Visibility)
          .switch(Repository)
          .join(Namespace, on=(Repository.namespace_user == Namespace.id))
          .where(Namespace.username == username, Visibility.name == 'private')
          .count())


def get_active_users():
  return User.select().where(User.organization == False, User.robot == False)


def get_active_user_count():
  return get_active_users().count()


def detach_external_login(user, service_name):
  try:
    service = LoginService.get(name=service_name)
  except LoginService.DoesNotExist:
    return

  FederatedLogin.delete().where(FederatedLogin.user == user,
                                FederatedLogin.service == service).execute()


def delete_user(user):
  # Delete any repositories under the user's namespace.
  for repo in list(Repository.select().where(Repository.namespace_user == user)):
    repository.purge_repository(user.username, repo.name)

  # Delete the user itself.
  user.delete_instance(recursive=True, delete_nullable=True)


def get_pull_credentials(robotname):
  try:
    robot = lookup_robot(robotname)
  except InvalidRobotException:
    return None

  try:
    login_info = FederatedLogin.get(user=robot)
  except FederatedLogin.DoesNotExist:
    return None

  return {
      'username': robot.username,
      'password': login_info.service_ident,
      'registry': '%s://%s/v1/' % (config.app_config['PREFERRED_URL_SCHEME'],
                                   config.app_config['SERVER_HOSTNAME']),
  }

def get_region_locations(user):
  """ Returns the locations defined as preferred storage for the given user. """
  query = UserRegion.select().join(ImageStorageLocation).where(UserRegion.user == user)
  return set([region.location.name for region in query])