Fix creating users from the admin panel while name changes disabled (#862)

* Fix creating users from the admin panel while name changes are disabled; clean up user & team schema validators
* Closes #832
* Coerce /api/v1/teams/<team_id> to /api/v1/teams/<int:team_id>
This commit is contained in:
Kevin Chung
2019-01-31 01:18:46 -05:00
committed by GitHub
parent 2935a76722
commit 2f49477465
7 changed files with 178 additions and 25 deletions

View File

@@ -73,7 +73,7 @@ class TeamList(Resource):
} }
@teams_namespace.route('/<team_id>') @teams_namespace.route('/<int:team_id>')
@teams_namespace.param('team_id', "Team ID") @teams_namespace.param('team_id', "Team ID")
class TeamPublic(Resource): class TeamPublic(Resource):
@check_account_visibility @check_account_visibility

View File

@@ -58,7 +58,7 @@ class TeamSchema(ma.ModelSchema):
if is_admin(): if is_admin():
team_id = int(data.get('id', 0)) team_id = int(data.get('id', 0))
if team_id: if team_id:
if existing_team.id != team_id: if existing_team and existing_team.id != team_id:
raise ValidationError('Team name has already been taken', field_names=['name']) raise ValidationError('Team name has already been taken', field_names=['name'])
else: else:
# If there's no Team ID it means that the admin is creating a team with no ID. # If there's no Team ID it means that the admin is creating a team with no ID.
@@ -83,18 +83,21 @@ class TeamSchema(ma.ModelSchema):
if email is None: if email is None:
return return
obj = Teams.query.filter_by(email=email).first() existing_team = Teams.query.filter_by(email=email).first()
if obj:
if is_admin(): if is_admin():
if data.get('id'): team_id = data.get('id')
target_user = Teams.query.filter_by(id=data['id']).first() if team_id:
else: if existing_team and existing_team.id != team_id:
target_user = get_current_team()
if target_user and obj.id != target_user.id:
raise ValidationError('Email address has already been used', field_names=['email']) raise ValidationError('Email address has already been used', field_names=['email'])
else: else:
if obj.id != get_current_team().id: if existing_team:
raise ValidationError('Email address has already been used', field_names=['email'])
else:
current_team = get_current_team()
if email == current_team.email:
return data
else:
if existing_team:
raise ValidationError('Email address has already been used', field_names=['email']) raise ValidationError('Email address has already been used', field_names=['email'])
@pre_load @pre_load

View File

@@ -66,11 +66,14 @@ class UserSchema(ma.ModelSchema):
return return
existing_user = Users.query.filter_by(name=name).first() existing_user = Users.query.filter_by(name=name).first()
if is_admin():
user_id = data.get('id') user_id = data.get('id')
if user_id:
if user_id and is_admin():
if existing_user and existing_user.id != user_id: if existing_user and existing_user.id != user_id:
raise ValidationError('User name has already been taken', field_names=['name']) raise ValidationError('User name has already been taken', field_names=['name'])
else:
if existing_user:
raise ValidationError('User name has already been taken', field_names=['name'])
else: else:
current_user = get_current_user() current_user = get_current_user()
if name == current_user.name: if name == current_user.name:
@@ -89,11 +92,15 @@ class UserSchema(ma.ModelSchema):
return return
existing_user = Users.query.filter_by(email=email).first() existing_user = Users.query.filter_by(email=email).first()
user_id = data.get('id')
if user_id and is_admin(): if is_admin():
user_id = data.get('id')
if user_id:
if existing_user and existing_user.id != user_id: if existing_user and existing_user.id != user_id:
raise ValidationError('Email address has already been used', field_names=['email']) raise ValidationError('Email address has already been used', field_names=['email'])
else:
if existing_user:
raise ValidationError('Email address has already been used', field_names=['email'])
else: else:
current_user = get_current_user() current_user = get_current_user()
if email == current_user.email: if email == current_user.email:

View File

@@ -2,7 +2,7 @@ from passlib.hash import bcrypt_sha256
def hash_password(plaintext): def hash_password(plaintext):
return bcrypt_sha256.encrypt(str(plaintext)) return bcrypt_sha256.hash(str(plaintext))
def verify_password(plaintext, ciphertext): def verify_password(plaintext, ciphertext):

View File

@@ -109,6 +109,50 @@ def test_api_teams_post_admin():
destroy_ctfd(app) destroy_ctfd(app)
def test_api_teams_post_admin_duplicate():
"""Test that admins can only create teams with unique information"""
app = create_ctfd(user_mode="teams")
with app.app_context():
gen_team(app.db, name='team1')
with login_as_user(app, 'admin') as client:
# Duplicate name
r = client.post(
'/api/v1/teams',
json={
"website": "https://ctfd.io",
"name": "team1",
"country": "TW",
"email": "team1@ctfd.io",
"affiliation": "team",
"password": "password"
}
)
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['name']
assert resp['success'] is False
assert Teams.query.count() == 1
# Duplicate email
r = client.post(
'/api/v1/teams',
json={
"website": "https://ctfd.io",
"name": "new_team",
"country": "TW",
"email": "team@ctfd.io",
"affiliation": "team",
"password": "password"
}
)
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['email']
assert resp['success'] is False
assert Teams.query.count() == 1
destroy_ctfd(app)
def test_api_team_get_public(): def test_api_team_get_public():
"""Can a user get /api/v1/team/<team_id> if teams are public""" """Can a user get /api/v1/team/<team_id> if teams are public"""
app = create_ctfd(user_mode="teams") app = create_ctfd(user_mode="teams")
@@ -184,6 +228,7 @@ def test_api_team_patch_admin():
with login_as_user(app, 'admin') as client: with login_as_user(app, 'admin') as client:
r = client.patch('/api/v1/teams/1', json={ r = client.patch('/api/v1/teams/1', json={
"name": "team_name", "name": "team_name",
"email": "team@ctfd.io",
"password": "password", "password": "password",
"affiliation": "changed" "affiliation": "changed"
}) })
@@ -191,7 +236,6 @@ def test_api_team_patch_admin():
assert r.status_code == 200 assert r.status_code == 200
assert r.get_json()['data']['affiliation'] == 'changed' assert r.get_json()['data']['affiliation'] == 'changed'
assert verify_password('password', team.password) assert verify_password('password', team.password)
destroy_ctfd(app) destroy_ctfd(app)

View File

@@ -118,6 +118,102 @@ def test_api_users_post_admin_with_attributes():
destroy_ctfd(app) destroy_ctfd(app)
def test_api_users_post_admin_duplicate_information():
"""Can an admin create a user with duplicate information"""
app = create_ctfd()
with app.app_context():
register_user(app)
with login_as_user(app, 'admin') as client:
# Duplicate email
r = client.post('/api/v1/users', json={
"name": "user2",
"email": "user@ctfd.io",
"password": "password"
})
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['email']
assert resp['success'] is False
assert Users.query.count() == 2
# Duplicate user
r = client.post('/api/v1/users', json={
"name": "user",
"email": "user2@ctfd.io",
"password": "password"
})
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['name']
assert resp['success'] is False
assert Users.query.count() == 2
destroy_ctfd(app)
def test_api_users_patch_admin_duplicate_information():
"""Can an admin modify a user with duplicate information"""
app = create_ctfd()
with app.app_context():
register_user(app, name="user1", email="user1@ctfd.io", password="password")
register_user(app, name="user2", email="user2@ctfd.io", password="password")
with login_as_user(app, 'admin') as client:
# Duplicate name
r = client.patch('/api/v1/users/1', json={
"name": "user2",
"email": "user@ctfd.io",
"password": "password"
})
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['name']
assert resp['success'] is False
# Duplicate email
r = client.patch('/api/v1/users/1', json={
"name": "user",
"email": "user2@ctfd.io",
"password": "password"
})
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['email']
assert resp['success'] is False
assert Users.query.count() == 3
destroy_ctfd(app)
def test_api_users_patch_duplicate_information():
"""Can a user modify their information to another user's"""
app = create_ctfd()
with app.app_context():
register_user(app, name="user1", email="user1@ctfd.io", password="password")
register_user(app, name="user2", email="user2@ctfd.io", password="password")
with login_as_user(app, 'user1') as client:
# Duplicate email
r = client.patch('/api/v1/users/me', json={
"name": "user2",
"email": "user@ctfd.io",
"password": "password"
})
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['name']
assert resp['success'] is False
# Duplicate user
r = client.patch('/api/v1/users/me', json={
"name": "user",
"email": "user2@ctfd.io",
"password": "password"
})
resp = r.get_json()
assert r.status_code == 400
assert resp['errors']['email']
assert resp['success'] is False
assert Users.query.count() == 3
destroy_ctfd(app)
def test_api_team_get_public(): def test_api_team_get_public():
"""Can a user get /api/v1/team/<user_id> if users are public""" """Can a user get /api/v1/team/<user_id> if users are public"""
app = create_ctfd() app = create_ctfd()

View File

@@ -98,7 +98,7 @@ def register_user(app, name="user", email="user@ctfd.io", password="password", r
assert sess['nonce'] assert sess['nonce']
def register_team(app, name="team", password="password"): def register_team(app, name="team", password="password", raise_for_error=True):
with app.app_context(): with app.app_context():
with app.test_client() as client: with app.test_client() as client:
r = client.get('/team') r = client.get('/team')
@@ -108,7 +108,10 @@ def register_team(app, name="team", password="password"):
"password": password, "password": password,
"nonce": sess.get('nonce') "nonce": sess.get('nonce')
} }
client.post('/teams/new', data=data) r = client.post('/teams/new', data=data)
if raise_for_error:
assert r.status_code == 302
return client
def login_as_user(app, name="user", password="password", raise_for_error=True): def login_as_user(app, name="user", password="password", raise_for_error=True):