server/users: harden password hashes
- Changed password setup to use libsodium and argon2id (regular SHA256 hashing for passwords is inadequate as modern GPU's can hash generate billions of hashes per second). - Added code to auto migrate old passwords to the new password_hash if the existing password_hash matches either of the legacy password generation schemes (SHA1 or SHA256). - Added migration to support new password_hash format length - Added column password_revision. This field will default to 0, which all passwords will have till they're updated. After that each password hash method has a revision.
This commit is contained in:
ReAnzu
rr-
parent
7519e071e7
commit
3f52aceca4
|
|
@ -11,3 +11,4 @@ scipy>=0.18.1
|
|||
elasticsearch>=5.0.0
|
||||
elasticsearch-dsl>=5.0.0
|
||||
scikit-image>=0.12
|
||||
pynacl>=1.2.1
|
||||
|
|
@ -1,7 +1,10 @@
|
|||
from typing import Tuple
|
||||
import hashlib
|
||||
import random
|
||||
from collections import OrderedDict
|
||||
from szurubooru import config, model, errors
|
||||
from nacl import pwhash
|
||||
from nacl.exceptions import InvalidkeyError
|
||||
from szurubooru import config, model, errors, db
|
||||
from szurubooru.func import util
|
||||
|
||||
|
||||
|
|
@ -16,22 +19,29 @@ RANK_MAP = OrderedDict([
|
|||
])
|
||||
|
||||
|
||||
def get_password_hash(salt: str, password: str) -> str:
|
||||
''' Retrieve new-style password hash. '''
|
||||
def get_password_hash(salt: str, password: str) -> Tuple[str, int]:
|
||||
''' Retrieve argon2id password hash. '''
|
||||
return pwhash.argon2id.str(
|
||||
(config.config['secret'] + salt + password).encode('utf8')
|
||||
).decode('utf8'), 3
|
||||
|
||||
|
||||
def get_sha256_legacy_password_hash(salt: str, password: str) -> Tuple[str, int]:
|
||||
''' Retrieve old-style sha256 password hash. '''
|
||||
digest = hashlib.sha256()
|
||||
digest.update(config.config['secret'].encode('utf8'))
|
||||
digest.update(salt.encode('utf8'))
|
||||
digest.update(password.encode('utf8'))
|
||||
return digest.hexdigest()
|
||||
return digest.hexdigest(), 2
|
||||
|
||||
|
||||
def get_legacy_password_hash(salt: str, password: str) -> str:
|
||||
''' Retrieve old-style password hash. '''
|
||||
def get_sha1_legacy_password_hash(salt: str, password: str) -> Tuple[str, int]:
|
||||
''' Retrieve old-style sha1 password hash. '''
|
||||
digest = hashlib.sha1()
|
||||
digest.update(b'1A2/$_4xVa')
|
||||
digest.update(salt.encode('utf8'))
|
||||
digest.update(password.encode('utf8'))
|
||||
return digest.hexdigest()
|
||||
return digest.hexdigest(), 1
|
||||
|
||||
|
||||
def create_password() -> str:
|
||||
|
|
@ -47,11 +57,25 @@ def create_password() -> str:
|
|||
def is_valid_password(user: model.User, password: str) -> bool:
|
||||
assert user
|
||||
salt, valid_hash = user.password_salt, user.password_hash
|
||||
possible_hashes = [
|
||||
get_password_hash(salt, password),
|
||||
get_legacy_password_hash(salt, password)
|
||||
]
|
||||
return valid_hash in possible_hashes
|
||||
|
||||
try:
|
||||
return pwhash.verify(
|
||||
user.password_hash.encode('utf8'),
|
||||
(config.config['secret'] + salt + password).encode('utf8'))
|
||||
except InvalidkeyError:
|
||||
possible_hashes = [
|
||||
get_sha256_legacy_password_hash(salt, password)[0],
|
||||
get_sha1_legacy_password_hash(salt, password)[0]
|
||||
]
|
||||
if valid_hash in possible_hashes:
|
||||
# Convert the user password hash to the new hash
|
||||
new_hash, revision = get_password_hash(salt, password)
|
||||
user.password_hash = new_hash
|
||||
user.password_revision = revision
|
||||
db.session.commit()
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def has_privilege(user: model.User, privilege_name: str) -> bool:
|
||||
|
|
|
|||
|
|
@ -243,7 +243,10 @@ def update_user_password(user: model.User, password: str) -> None:
|
|||
raise InvalidPasswordError(
|
||||
'Password must satisfy regex %r.' % password_regex)
|
||||
user.password_salt = auth.create_password()
|
||||
user.password_hash = auth.get_password_hash(user.password_salt, password)
|
||||
password_hash, revision = auth.get_password_hash(
|
||||
user.password_salt, password)
|
||||
user.password_hash = password_hash
|
||||
user.password_revision = revision
|
||||
|
||||
|
||||
def update_user_email(user: model.User, email: str) -> None:
|
||||
|
|
@ -308,5 +311,8 @@ def reset_user_password(user: model.User) -> str:
|
|||
assert user
|
||||
password = auth.create_password()
|
||||
user.password_salt = auth.create_password()
|
||||
user.password_hash = auth.get_password_hash(user.password_salt, password)
|
||||
password_hash, revision = auth.get_password_hash(
|
||||
user.password_salt, password)
|
||||
user.password_hash = password_hash
|
||||
user.password_revision = revision
|
||||
return password
|
||||
|
|
|
|||
|
|
@ -35,7 +35,10 @@ def run_migrations_offline():
|
|||
'''
|
||||
url = alembic_config.get_main_option('sqlalchemy.url')
|
||||
alembic.context.configure(
|
||||
url=url, target_metadata=target_metadata, literal_binds=True)
|
||||
url=url,
|
||||
target_metadata=target_metadata,
|
||||
literal_binds=True,
|
||||
compare_type=True)
|
||||
|
||||
with alembic.context.begin_transaction():
|
||||
alembic.context.run_migrations()
|
||||
|
|
@ -56,7 +59,8 @@ def run_migrations_online():
|
|||
with connectable.connect() as connection:
|
||||
alembic.context.configure(
|
||||
connection=connection,
|
||||
target_metadata=target_metadata)
|
||||
target_metadata=target_metadata,
|
||||
compare_type=True)
|
||||
|
||||
with alembic.context.begin_transaction():
|
||||
alembic.context.run_migrations()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,89 @@
|
|||
'''
|
||||
Alter the password_hash field to work with larger output.
|
||||
Particularly libsodium output for greater password security.
|
||||
|
||||
Revision ID: 9ef1a1643c2a
|
||||
Created at: 2018-02-24 23:00:32.848575
|
||||
'''
|
||||
|
||||
import sqlalchemy as sa
|
||||
import sqlalchemy.ext.declarative
|
||||
import sqlalchemy.orm.session
|
||||
from alembic import op
|
||||
|
||||
|
||||
revision = '9ef1a1643c2a'
|
||||
down_revision = '02ef5f73f4ab'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
Base = sa.ext.declarative.declarative_base()
|
||||
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = 'user'
|
||||
|
||||
AVATAR_GRAVATAR = 'gravatar'
|
||||
|
||||
user_id = sa.Column('id', sa.Integer, primary_key=True)
|
||||
creation_time = sa.Column('creation_time', sa.DateTime, nullable=False)
|
||||
last_login_time = sa.Column('last_login_time', sa.DateTime)
|
||||
version = sa.Column('version', sa.Integer, default=1, nullable=False)
|
||||
name = sa.Column('name', sa.Unicode(50), nullable=False, unique=True)
|
||||
password_hash = sa.Column('password_hash', sa.Unicode(128), nullable=False)
|
||||
password_salt = sa.Column('password_salt', sa.Unicode(32))
|
||||
password_revision = sa.Column(
|
||||
'password_revision', sa.SmallInteger, default=0, nullable=False)
|
||||
email = sa.Column('email', sa.Unicode(64), nullable=True)
|
||||
rank = sa.Column('rank', sa.Unicode(32), nullable=False)
|
||||
avatar_style = sa.Column(
|
||||
'avatar_style', sa.Unicode(32), nullable=False,
|
||||
default=AVATAR_GRAVATAR)
|
||||
|
||||
__mapper_args__ = {
|
||||
'version_id_col': version,
|
||||
'version_id_generator': False,
|
||||
}
|
||||
|
||||
|
||||
def upgrade():
|
||||
op.alter_column(
|
||||
'user',
|
||||
'password_hash',
|
||||
existing_type=sa.VARCHAR(length=64),
|
||||
type_=sa.Unicode(length=128),
|
||||
existing_nullable=False)
|
||||
op.add_column('user', sa.Column(
|
||||
'password_revision',
|
||||
sa.SmallInteger(),
|
||||
nullable=True,
|
||||
default=0))
|
||||
|
||||
session = sa.orm.session.Session(bind=op.get_bind())
|
||||
if session.query(User).count() >= 0:
|
||||
for user in session.query(User).all():
|
||||
password_hash_length = len(user.password_hash)
|
||||
if password_hash_length == 40:
|
||||
user.password_revision = 1
|
||||
elif password_hash_length == 64:
|
||||
user.password_revision = 2
|
||||
else:
|
||||
user.password_revision = 0
|
||||
session.flush()
|
||||
session.commit()
|
||||
|
||||
op.alter_column(
|
||||
'user',
|
||||
'password_revision',
|
||||
existing_nullable=True,
|
||||
nullable=False)
|
||||
|
||||
|
||||
def downgrade():
|
||||
op.alter_column(
|
||||
'user',
|
||||
'password_hash',
|
||||
existing_type=sa.Unicode(length=128),
|
||||
type_=sa.VARCHAR(length=64),
|
||||
existing_nullable=False)
|
||||
op.drop_column('user', 'password_revision')
|
||||
|
|
@ -23,8 +23,10 @@ class User(Base):
|
|||
last_login_time = sa.Column('last_login_time', sa.DateTime)
|
||||
version = sa.Column('version', sa.Integer, default=1, nullable=False)
|
||||
name = sa.Column('name', sa.Unicode(50), nullable=False, unique=True)
|
||||
password_hash = sa.Column('password_hash', sa.Unicode(64), nullable=False)
|
||||
password_hash = sa.Column('password_hash', sa.Unicode(128), nullable=False)
|
||||
password_salt = sa.Column('password_salt', sa.Unicode(32))
|
||||
password_revision = sa.Column(
|
||||
'password_revision', sa.SmallInteger, default=0, nullable=False)
|
||||
email = sa.Column('email', sa.Unicode(64), nullable=True)
|
||||
rank = sa.Column('rank', sa.Unicode(32), nullable=False)
|
||||
avatar_style = sa.Column(
|
||||
|
|
|
|||
|
|
@ -115,11 +115,16 @@ def config_injector():
|
|||
|
||||
@pytest.fixture
|
||||
def user_factory():
|
||||
def factory(name=None, rank=model.User.RANK_REGULAR, email='dummy'):
|
||||
def factory(
|
||||
name=None,
|
||||
rank=model.User.RANK_REGULAR,
|
||||
email='dummy',
|
||||
password_salt=None,
|
||||
password_hash=None):
|
||||
user = model.User()
|
||||
user.name = name or get_unique_name()
|
||||
user.password_salt = 'dummy'
|
||||
user.password_hash = 'dummy'
|
||||
user.password_salt = password_salt or 'dummy'
|
||||
user.password_hash = password_hash or 'dummy'
|
||||
user.email = email
|
||||
user.rank = rank
|
||||
user.creation_time = datetime(1997, 1, 1)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,43 @@
|
|||
from szurubooru.func import auth
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def inject_config(config_injector):
|
||||
config_injector({'secret': 'testSecret'})
|
||||
|
||||
|
||||
def test_get_password_hash():
|
||||
salt, password = ('testSalt', 'pass')
|
||||
result, revision = auth.get_password_hash(salt, password)
|
||||
assert result
|
||||
assert revision == 3
|
||||
hash_parts = list(
|
||||
filter(lambda e: e is not None and e != '', result.split('$')))
|
||||
assert len(hash_parts) == 5
|
||||
assert hash_parts[0] == 'argon2id'
|
||||
|
||||
|
||||
def test_get_sha256_legacy_password_hash():
|
||||
salt, password = ('testSalt', 'pass')
|
||||
result, revision = auth.get_sha256_legacy_password_hash(salt, password)
|
||||
hash = '2031ac9631353ac9303719a7f808a24f79aa1d71712c98523e4bb4cce579428a'
|
||||
assert result == hash
|
||||
assert revision == 2
|
||||
|
||||
|
||||
def test_get_sha1_legacy_password_hash():
|
||||
salt, password = ('testSalt', 'pass')
|
||||
result, revision = auth.get_sha1_legacy_password_hash(salt, password)
|
||||
assert result == '1eb1f953d9be303a1b54627e903e6124cfb1245b'
|
||||
assert revision == 1
|
||||
|
||||
|
||||
def test_is_valid_password_auto_upgrades_user_password_hash(user_factory):
|
||||
salt, password = ('testSalt', 'pass')
|
||||
hash, revision = auth.get_sha256_legacy_password_hash(salt, password)
|
||||
user = user_factory(password_salt=salt, password_hash=hash)
|
||||
result = auth.is_valid_password(user, password)
|
||||
assert result is True
|
||||
assert user.password_hash != hash
|
||||
assert user.password_revision > revision
|
||||
|
|
@ -320,10 +320,11 @@ def test_update_user_password(user_factory, config_injector):
|
|||
with patch('szurubooru.func.auth.create_password'), \
|
||||
patch('szurubooru.func.auth.get_password_hash'):
|
||||
auth.create_password.return_value = 'salt'
|
||||
auth.get_password_hash.return_value = 'hash'
|
||||
auth.get_password_hash.return_value = ('hash', 3)
|
||||
users.update_user_password(user, 'a')
|
||||
assert user.password_salt == 'salt'
|
||||
assert user.password_hash == 'hash'
|
||||
assert user.password_revision == 3
|
||||
|
||||
|
||||
def test_update_user_email_with_too_long_string(user_factory):
|
||||
|
|
@ -447,7 +448,8 @@ def test_reset_user_password(user_factory):
|
|||
patch('szurubooru.func.auth.get_password_hash'):
|
||||
user = user_factory()
|
||||
auth.create_password.return_value = 'salt'
|
||||
auth.get_password_hash.return_value = 'hash'
|
||||
auth.get_password_hash.return_value = ('hash', 3)
|
||||
users.reset_user_password(user)
|
||||
assert user.password_salt == 'salt'
|
||||
assert user.password_hash == 'hash'
|
||||
assert user.password_revision == 3
|
||||
|
|
|
|||
Loading…
Reference in New Issue