837 lines
27 KiB
Python
837 lines
27 KiB
Python
import bcrypt
|
|
import logging
|
|
import json
|
|
import uuid
|
|
|
|
from peewee import JOIN_LEFT_OUTER, IntegrityError, fn
|
|
from uuid import uuid4
|
|
from datetime import datetime, timedelta
|
|
from enum import Enum
|
|
|
|
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, QueueItem, TeamMemberInvite,
|
|
ServiceKeyApproval, OAuthApplication, RepositoryBuildTrigger,
|
|
UserPromptKind, UserPrompt, UserPromptTypes)
|
|
from data.model import (DataModelException, InvalidPasswordException, InvalidRobotException,
|
|
InvalidUsernameException, InvalidEmailAddressException,
|
|
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 create_user(username, password, email, auto_verify=False, email_required=True, prompts=tuple()):
|
|
""" Creates a regular user, if allowed. """
|
|
if not validate_password(password):
|
|
raise InvalidPasswordException(INVALID_PASSWORD_MESSAGE)
|
|
|
|
created = create_user_noverify(username, email, email_required=email_required, prompts=prompts)
|
|
created.password_hash = hash_password(password)
|
|
created.verified = auto_verify
|
|
created.save()
|
|
|
|
return created
|
|
|
|
|
|
def create_user_noverify(username, email, email_required=True, prompts=tuple()):
|
|
if email_required:
|
|
if not validate_email(email):
|
|
raise InvalidEmailAddressException('Invalid email address: %s' % email)
|
|
else:
|
|
# If email addresses are not required and none was specified, then we just use a unique
|
|
# ID to ensure that the database consistency check remains intact.
|
|
email = email or str(uuid.uuid4())
|
|
|
|
(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:
|
|
new_user = User.create(username=username, email=email)
|
|
for prompt in prompts:
|
|
create_user_prompt(new_user, prompt)
|
|
|
|
return new_user
|
|
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 get_default_user_prompts(features):
|
|
prompts = set()
|
|
if features.USER_METADATA:
|
|
prompts.add(UserPromptTypes.ENTER_NAME)
|
|
prompts.add(UserPromptTypes.ENTER_COMPANY)
|
|
|
|
return prompts
|
|
|
|
|
|
def has_user_prompts(user):
|
|
try:
|
|
UserPrompt.select().where(UserPrompt.user == user).get()
|
|
return True
|
|
except UserPrompt.DoesNotExist:
|
|
return False
|
|
|
|
|
|
def has_user_prompt(user, prompt_name):
|
|
prompt_kind = UserPromptKind.get(name=prompt_name)
|
|
|
|
try:
|
|
UserPrompt.get(user=user, kind=prompt_kind)
|
|
return True
|
|
except UserPrompt.DoesNotExist:
|
|
return False
|
|
|
|
|
|
def create_user_prompt(user, prompt_name):
|
|
prompt_kind = UserPromptKind.get(name=prompt_name)
|
|
return UserPrompt.create(user=user, kind=prompt_kind)
|
|
|
|
|
|
def remove_user_prompt(user, prompt_name):
|
|
prompt_kind = UserPromptKind.get(name=prompt_name)
|
|
UserPrompt.delete().where(UserPrompt.user == user, UserPrompt.kind == prompt_kind).execute()
|
|
|
|
|
|
def get_user_prompts(user):
|
|
query = UserPrompt.select().where(UserPrompt.user == user).join(UserPromptKind)
|
|
return [prompt.kind.name for prompt in query]
|
|
|
|
|
|
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()
|
|
|
|
# Remove any prompts for username.
|
|
remove_user_prompt(user, 'confirm_username')
|
|
|
|
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 update_user_metadata(user, given_name=None, family_name=None, company=None):
|
|
""" Updates the metadata associated with the user, including his/her name and company. """
|
|
with db_transaction():
|
|
user.given_name = given_name or user.given_name
|
|
user.family_name = family_name or user.family_name
|
|
user.company = company or user.company
|
|
user.save()
|
|
|
|
# Remove any prompts associated with the user's metadata being needed.
|
|
remove_user_prompt(user, UserPromptTypes.ENTER_NAME)
|
|
remove_user_prompt(user, UserPromptTypes.ENTER_COMPANY)
|
|
|
|
|
|
def create_federated_user(username, email, service_name, service_ident,
|
|
set_password_notification, metadata={},
|
|
email_required=True, prompts=tuple()):
|
|
prompts = set(prompts)
|
|
prompts.add(UserPromptTypes.CONFIRM_USERNAME)
|
|
|
|
new_user = create_user_noverify(username, email, email_required=email_required, prompts=prompts)
|
|
new_user.verified = True
|
|
new_user.save()
|
|
|
|
service = LoginService.get(LoginService.name == service_name)
|
|
FederatedLogin.create(user=new_user, service=service,
|
|
service_ident=service_ident,
|
|
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_ident, metadata={}):
|
|
service = LoginService.get(LoginService.name == service_name)
|
|
FederatedLogin.create(user=user, service=service, service_ident=service_ident,
|
|
metadata_json=json.dumps(metadata))
|
|
return user
|
|
|
|
|
|
def verify_federated_login(service_name, service_ident):
|
|
try:
|
|
found = (FederatedLogin
|
|
.select(FederatedLogin, User)
|
|
.join(LoginService)
|
|
.switch(FederatedLogin).join(User)
|
|
.where(FederatedLogin.service_ident == service_ident, 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
|
|
if not user.verified:
|
|
user.verified = True
|
|
user.save()
|
|
|
|
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, limit=20):
|
|
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.id, User.username, User.email, User.robot)
|
|
.group_by(User.id, User.username, User.email, User.robot)
|
|
.where(direct_user_query))
|
|
|
|
if organization:
|
|
query = (query
|
|
.select(User.id, 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.id = args[0]
|
|
self.username = args[1]
|
|
self.email = args[2]
|
|
self.robot = args[3]
|
|
|
|
if organization:
|
|
self.is_org_member = (args[3] != None)
|
|
else:
|
|
self.is_org_member = None
|
|
|
|
return (MatchingUserResult(*args) for args in query.tuples().limit(limit))
|
|
|
|
|
|
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 get_robot_count():
|
|
return User.select().where(User.robot == True).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 get_solely_admined_organizations(user_obj):
|
|
""" Returns the organizations admined solely by the given user. """
|
|
orgs = (User.select()
|
|
.where(User.organization == True)
|
|
.join(Team)
|
|
.join(TeamRole)
|
|
.where(TeamRole.name == 'admin')
|
|
.switch(Team)
|
|
.join(TeamMember)
|
|
.where(TeamMember.user == user_obj)
|
|
.distinct())
|
|
|
|
# Filter to organizations where the user is the sole admin.
|
|
solely_admined = []
|
|
for org in orgs:
|
|
admin_user_count = (TeamMember.select()
|
|
.join(Team)
|
|
.join(TeamRole)
|
|
.where(Team.organization == org, TeamRole.name == 'admin')
|
|
.switch(TeamMember)
|
|
.join(User)
|
|
.where(User.robot == False)
|
|
.distinct()
|
|
.count())
|
|
|
|
if admin_user_count == 1:
|
|
solely_admined.append(org)
|
|
|
|
return solely_admined
|
|
|
|
|
|
def delete_user(user, queues, force=False):
|
|
if not force and not user.organization:
|
|
# Ensure that the user is not the sole admin for any organizations. If so, then the user
|
|
# cannot be deleted before those organizations are deleted or reassigned.
|
|
organizations = get_solely_admined_organizations(user)
|
|
if len(organizations) > 0:
|
|
message = 'Cannot delete %s as you are the only admin for organizations: ' % user.username
|
|
for index, org in enumerate(organizations):
|
|
if index > 0:
|
|
message = message + ', '
|
|
|
|
message = message + org.username
|
|
|
|
raise DataModelException(message)
|
|
|
|
# Delete all queue items for the user.
|
|
for queue in queues:
|
|
queue.delete_namespaced_items(user.username)
|
|
|
|
# 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)
|
|
|
|
if user.organization:
|
|
# Delete the organization's teams.
|
|
for team in Team.select().where(Team.organization == user):
|
|
team.delete_instance(recursive=True)
|
|
|
|
# Delete any OAuth approvals and tokens associated with the user.
|
|
for app in OAuthApplication.select().where(OAuthApplication.organization == user):
|
|
app.delete_instance(recursive=True)
|
|
else:
|
|
# Remove the user from any teams in which they are a member.
|
|
TeamMember.delete().where(TeamMember.user == user).execute()
|
|
|
|
# Delete any repository buildtriggers where the user is the connected user.
|
|
triggers = RepositoryBuildTrigger.select().where(RepositoryBuildTrigger.connected_user == user)
|
|
for trigger in triggers:
|
|
trigger.delete_instance(recursive=True, delete_nullable=False)
|
|
|
|
# Null out any service key approvals. We technically lose information here, but its better than
|
|
# falling and only occurs if a superuser is being deleted.
|
|
ServiceKeyApproval.update(approver=None).where(ServiceKeyApproval.approver == user).execute()
|
|
|
|
# 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])
|
|
|
|
def get_federated_logins(user_ids, service_name):
|
|
""" Returns all federated logins for the given user ids under the given external service. """
|
|
if not user_ids:
|
|
return []
|
|
|
|
return (FederatedLogin
|
|
.select()
|
|
.join(User)
|
|
.switch(FederatedLogin)
|
|
.join(LoginService)
|
|
.where(FederatedLogin.user << user_ids,
|
|
LoginService.name == service_name))
|