diff options
Diffstat (limited to 'gn3/auth/authentication/users.py')
-rw-r--r-- | gn3/auth/authentication/users.py | 128 |
1 files changed, 0 insertions, 128 deletions
diff --git a/gn3/auth/authentication/users.py b/gn3/auth/authentication/users.py deleted file mode 100644 index 0e72ed2..0000000 --- a/gn3/auth/authentication/users.py +++ /dev/null @@ -1,128 +0,0 @@ -"""User-specific code and data structures.""" -from uuid import UUID, uuid4 -from typing import Any, Tuple, NamedTuple - -from argon2 import PasswordHasher -from argon2.exceptions import VerifyMismatchError - -from gn3.auth import db -from gn3.auth.authorisation.errors import NotFoundError - -class User(NamedTuple): - """Class representing a user.""" - user_id: UUID - email: str - name: str - - def get_user_id(self): - """Return the user's UUID. Mostly for use with Authlib.""" - return self.user_id - - def dictify(self) -> dict[str, Any]: - """Return a dict representation of `User` objects.""" - return {"user_id": self.user_id, "email": self.email, "name": self.name} - -DUMMY_USER = User(user_id=UUID("a391cf60-e8b7-4294-bd22-ddbbda4b3530"), - email="gn3@dummy.user", - name="Dummy user to use as placeholder") - -def user_by_email(conn: db.DbConnection, email: str) -> User: - """Retrieve user from database by their email address""" - with db.cursor(conn) as cursor: - cursor.execute("SELECT * FROM users WHERE email=?", (email,)) - row = cursor.fetchone() - - if row: - return User(UUID(row["user_id"]), row["email"], row["name"]) - - raise NotFoundError(f"Could not find user with email {email}") - -def user_by_id(conn: db.DbConnection, user_id: UUID) -> User: - """Retrieve user from database by their user id""" - with db.cursor(conn) as cursor: - cursor.execute("SELECT * FROM users WHERE user_id=?", (str(user_id),)) - row = cursor.fetchone() - - if row: - return User(UUID(row["user_id"]), row["email"], row["name"]) - - raise NotFoundError(f"Could not find user with ID {user_id}") - -def same_password(password: str, hashed: str) -> bool: - """Check that `raw_password` is hashed to `hash`""" - try: - return hasher().verify(hashed, password) - except VerifyMismatchError as _vme: - return False - -def valid_login(conn: db.DbConnection, user: User, password: str) -> bool: - """Check the validity of the provided credentials for login.""" - with db.cursor(conn) as cursor: - cursor.execute( - ("SELECT * FROM users LEFT JOIN user_credentials " - "ON users.user_id=user_credentials.user_id " - "WHERE users.user_id=?"), - (str(user.user_id),)) - row = cursor.fetchone() - - if row is None: - return False - - return same_password(password, row["password"]) - -def save_user(cursor: db.DbCursor, email: str, name: str) -> User: - """ - Create and persist a user. - - The user creation could be done during a transaction, therefore the function - takes a cursor object rather than a connection. - - The newly created and persisted user is then returned. - """ - user_id = uuid4() - cursor.execute("INSERT INTO users VALUES (?, ?, ?)", - (str(user_id), email, name)) - return User(user_id, email, name) - -def hasher(): - """Retrieve PasswordHasher object""" - # TODO: Maybe tune the parameters here... - # Tuneable Parameters: - # - time_cost (default: 2) - # - memory_cost (default: 102400) - # - parallelism (default: 8) - # - hash_len (default: 16) - # - salt_len (default: 16) - # - encoding (default: 'utf-8') - # - type (default: <Type.ID: 2>) - return PasswordHasher() - -def hash_password(password): - """Hash the password.""" - return hasher().hash(password) - -def set_user_password( - cursor: db.DbCursor, user: User, password: str) -> Tuple[User, bytes]: - """Set the given user's password in the database.""" - hashed_password = hash_password(password) - cursor.execute( - ("INSERT INTO user_credentials VALUES (:user_id, :hash) " - "ON CONFLICT (user_id) DO UPDATE SET password=:hash"), - {"user_id": str(user.user_id), "hash": hashed_password}) - return user, hashed_password - -def users(conn: db.DbConnection, - ids: tuple[UUID, ...] = tuple()) -> tuple[User, ...]: - """ - Fetch all users with the given `ids`. If `ids` is empty, return ALL users. - """ - params = ", ".join(["?"] * len(ids)) - with db.cursor(conn) as cursor: - query = "SELECT * FROM users" + ( - f" WHERE user_id IN ({params})" - if len(ids) > 0 else "") - print(query) - cursor.execute(query, tuple(str(the_id) for the_id in ids)) - return tuple(User(UUID(row["user_id"]), row["email"], row["name"]) - for row in cursor.fetchall()) - return tuple() |