feat: PasswordHash field type (#452)

Implements a PasswordHash field type with multiple supported backends.

Includes built-in backends for:
- `passlib`
- `argon2`
- `pwdlib`

Author: cofin

README.md

@@ -39,6 +39,8 @@ offering:
 - Optimized JSON types including a custom JSON type for Oracle
 - Integrated support for UUID6 and UUID7 using [`uuid-utils`](https://github.com/aminalaee/uuid-utils) (install with the `uuid` extra)
 - Integrated support for Nano ID using [`fastnanoid`](https://github.com/oliverlambson/fastnanoid) (install with the `nanoid` extra)
+- Custom encrypted text type with multiple backend support including [`pgcrypto`](https://www.postgresql.org/docs/current/pgcrypto.html) for PostgreSQL and the Fernet implementation from [`cryptography`](https://cryptography.io/en/latest/) for other databases
+- Custom password hashing type with multiple backend support including [`Argon2`](https://github.com/P-H-C/phc-winner-argon2), [`Passlib`](https://passlib.readthedocs.io/en/stable/), and [`Pwdlib`](https://pwdlib.readthedocs.io/en/stable/) with automatic salt generation
 - Pre-configured base classes with audit columns UUID or Big Integer primary keys and
   a [sentinel column](https://docs.sqlalchemy.org/en/20/core/connections.html#configuring-sentinel-columns).
 - Synchronous and asynchronous repositories featuring:

advanced_alchemy/alembic/templates/asyncio/script.py.mako

@@ -11,7 +11,7 @@ from typing import TYPE_CHECKING
 
 import sqlalchemy as sa
 from alembic import op
-from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC, StoredObject
+from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC, StoredObject, PasswordHash
 from sqlalchemy import Text  # noqa: F401
 ${imports if imports else ""}
 if TYPE_CHECKING:

advanced_alchemy/alembic/templates/sync/script.py.mako

@@ -11,7 +11,7 @@ from typing import TYPE_CHECKING
 
 import sqlalchemy as sa
 from alembic import op
-from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC, StoredObject
+from advanced_alchemy.types import EncryptedString, EncryptedText, GUID, ORA_JSONB, DateTimeUTC, StoredObject, PasswordHash
 from sqlalchemy import Text  # noqa: F401
 ${imports if imports else ""}
 if TYPE_CHECKING:

advanced_alchemy/types/__init__.py

@@ -1,13 +1,12 @@
 """SQLAlchemy custom types for use with the ORM."""
 
-from advanced_alchemy.types import file_object
+from advanced_alchemy.types import encrypted_string, file_object, password_hash
 from advanced_alchemy.types.datetime import DateTimeUTC
 from advanced_alchemy.types.encrypted_string import (
     EncryptedString,
     EncryptedText,
     EncryptionBackend,
     FernetBackend,
-    PGCryptoBackend,
 )
 from advanced_alchemy.types.file_object import (
     FileObject,
@@ -22,6 +21,7 @@
 from advanced_alchemy.types.identity import BigIntIdentity
 from advanced_alchemy.types.json import ORA_JSONB, JsonB
 from advanced_alchemy.types.mutables import MutableList
+from advanced_alchemy.types.password_hash.base import HashedPassword, PasswordHash
 
 __all__ = (
     "GUID",
@@ -36,13 +36,16 @@
     "FernetBackend",
     "FileObject",
     "FileObjectList",
+    "HashedPassword",
     "JsonB",
     "MutableList",
-    "PGCryptoBackend",
+    "PasswordHash",
     "StorageBackend",
     "StorageBackendT",
     "StorageRegistry",
     "StoredObject",
+    "encrypted_string",
     "file_object",
+    "password_hash",
     "storages",
 )

advanced_alchemy/types/password_hash/__init__.py

@@ -0,0 +1,64 @@
+"""Argon2 Hashing Backend using argon2-cffi."""
+
+from typing import TYPE_CHECKING, Any, Union
+
+from advanced_alchemy.types.password_hash.base import HashingBackend
+
+if TYPE_CHECKING:
+    from sqlalchemy import BinaryExpression, ColumnElement
+
+from argon2 import PasswordHasher as Argon2PasswordHasher  # pyright: ignore
+from argon2.exceptions import InvalidHash, VerifyMismatchError  # pyright: ignore
+
+
+class Argon2Hasher(HashingBackend):
+    """Hashing backend using Argon2 via the argon2-cffi library."""
+
+    def __init__(self, **kwargs: Any) -> None:
+        """Initialize Argon2Backend.
+
+        Args:
+            **kwargs: Optional keyword arguments to pass to the argon2.PasswordHasher constructor.
+                      See argon2-cffi documentation for available parameters (e.g., time_cost,
+                      memory_cost, parallelism, hash_len, salt_len, type).
+        """
+        self.hasher = Argon2PasswordHasher(**kwargs)  # pyright: ignore
+
+    def hash(self, value: "Union[str, bytes]") -> str:
+        """Hash the password using Argon2.
+
+        Args:
+            value: The plain text password (will be encoded to UTF-8 if string).
+
+        Returns:
+            The Argon2 hash string.
+        """
+        return self.hasher.hash(self._ensure_bytes(value))
+
+    def verify(self, plain: "Union[str, bytes]", hashed: str) -> bool:
+        """Verify a plain text password against an Argon2 hash.
+
+        Args:
+            plain: The plain text password (will be encoded to UTF-8 if string).
+            hashed: The Argon2 hash string to verify against.
+
+        Returns:
+            True if the password matches the hash, False otherwise.
+        """
+        try:
+            self.hasher.verify(hashed, self._ensure_bytes(plain))
+
+        except (VerifyMismatchError, InvalidHash):
+            return False
+        except Exception:  # noqa: BLE001
+            return False
+        return True
+
+    def compare_expression(self, column: "ColumnElement[str]", plain: "Union[str, bytes]") -> "BinaryExpression[bool]":
+        """Direct SQL comparison is not supported for Argon2.
+
+        Raises:
+            NotImplementedError: Always raised.
+        """
+        msg = "Argon2Hasher does not support direct SQL comparison."
+        raise NotImplementedError(msg)

advanced_alchemy/types/password_hash/argon2.py

@@ -0,0 +1,174 @@
+"""Base classes for password hashing backends."""
+
+import abc
+from typing import Any, Union, cast
+
+from sqlalchemy import BinaryExpression, ColumnElement, FunctionElement, String, TypeDecorator
+
+
+class HashingBackend(abc.ABC):
+    """Abstract base class for password hashing backends.
+
+    This class defines the interface that all password hashing backends must implement.
+    Concrete implementations should provide the actual hashing and verification logic.
+    """
+
+    @staticmethod
+    def _ensure_bytes(value: Union[str, bytes]) -> bytes:
+        if isinstance(value, str):
+            return value.encode("utf-8")
+        return value
+
+    @abc.abstractmethod
+    def hash(self, value: "Union[str, bytes]") -> "Union[str, Any]":
+        """Hash the given value.
+
+        Args:
+            value: The plain text value to hash.
+
+        Returns:
+            Either a string (the hash) or a SQLAlchemy SQL expression for DB-side hashing.
+        """
+
+    @abc.abstractmethod
+    def verify(self, plain: "Union[str, bytes]", hashed: str) -> bool:
+        """Verify a plain text value against a hash.
+
+        Args:
+            plain: The plain text value to verify.
+            hashed: The hash to verify against.
+
+        Returns:
+            True if the plain text matches the hash, False otherwise.
+        """
+
+    @abc.abstractmethod
+    def compare_expression(self, column: "ColumnElement[str]", plain: "Union[str, bytes]") -> "BinaryExpression[bool]":
+        """Generate a SQLAlchemy expression for comparing a column with a plain text value.
+
+        Args:
+            column: The SQLAlchemy column to compare.
+            plain: The plain text value to compare against.
+
+        Returns:
+            A SQLAlchemy binary expression for the comparison.
+        """
+
+
+class HashedPassword:
+    """Wrapper class for a hashed password.
+
+    This class holds the hash string and provides a method to verify a plain text password against it.
+    """
+
+    def __hash__(self) -> int:
+        return hash(self.hash_string)
+
+    def __init__(self, hash_string: str, backend: "HashingBackend") -> None:
+        """Initialize a HashedPassword object.
+
+        Args:
+            hash_string: The hash string.
+            backend: The hashing backend to use for verification.
+        """
+        self.hash_string = hash_string
+        self.backend = backend
+
+    def verify(self, plain_password: "Union[str, bytes]") -> bool:
+        """Verify a plain text password against this hash.
+
+        Args:
+            plain_password: The plain text password to verify.
+
+        Returns:
+            True if the password matches the hash, False otherwise.
+        """
+        return self.backend.verify(plain_password, self.hash_string)
+
+
+class PasswordHash(TypeDecorator[str]):
+    """SQLAlchemy TypeDecorator for storing hashed passwords in a database.
+
+    This type provides transparent hashing of password values using the specified backend.
+    It extends :class:`sqlalchemy.types.TypeDecorator` and implements String as its underlying type.
+    """
+
+    impl = String
+    cache_ok = True
+
+    def __init__(self, backend: "HashingBackend", length: int = 128) -> None:
+        """Initialize the PasswordHash TypeDecorator.
+
+        Args:
+            backend: The hashing backend class to use
+            length: The maximum length of the hash string. Defaults to 128.
+        """
+        self.length = length
+        super().__init__(length=length)
+        self.backend = backend
+
+    @property
+    def python_type(self) -> "type[str]":
+        """Returns the Python type for this type decorator.
+
+        Returns:
+            The Python string type.
+        """
+        return str
+
+    def process_bind_param(self, value: Any, dialect: Any) -> "Union[str, FunctionElement[str], None]":
+        """Process the value before binding it to the SQL statement.
+
+        This method hashes the value using the specified backend.
+        If the backend returns a SQLAlchemy FunctionElement (for DB-side hashing),
+        it is returned directly. Otherwise, the hashed string is returned.
+
+        Args:
+            value: The value to process.
+            dialect: The SQLAlchemy dialect.
+
+        Returns:
+            The hashed string, a SQLAlchemy FunctionElement, or None.
+        """
+        if value is None:
+            return value
+
+        hashed_value = self.backend.hash(value)
+
+        # Check if the backend returned a SQL function for DB-side hashing
+        if isinstance(hashed_value, FunctionElement):
+            return cast("FunctionElement[str]", hashed_value)
+
+        # Otherwise, assume it's a string or HashedPassword object (convert to string)
+        return str(hashed_value)
+
+    def process_result_value(self, value: Any, dialect: Any) -> "Union[HashedPassword, None]":  # type: ignore[override]
+        """Process the value after retrieving it from the database.
+
+        This method wraps the hash string in a HashedPassword object.
+
+        Args:
+            value: The value to process.
+            dialect: The SQLAlchemy dialect.
+
+        Returns:
+            A HashedPassword object or None if the input is None.
+        """
+        if value is None:
+            return value
+        # Ensure the retrieved value is a string before passing to HashedPassword
+        return HashedPassword(str(value), self.backend)
+
+    def compare_value(
+        self, column: "ColumnElement[str]", plain_password: "Union[str, bytes]"
+    ) -> "BinaryExpression[bool]":
+        """Generate a SQLAlchemy expression for comparing a column with a plain text password.
+
+        Args:
+            column: The SQLAlchemy column to compare.
+            plain_password: The plain text password to compare against.
+
+        Returns:
+            A SQLAlchemy binary expression for the comparison.
+        """
+        return self.backend.compare_expression(column, plain_password)

advanced_alchemy/types/password_hash/base.py

@@ -0,0 +1,62 @@
+"""Passlib Hashing Backend."""
+
+from typing import TYPE_CHECKING, Any, Union
+
+from passlib.context import CryptContext  # pyright: ignore
+
+from advanced_alchemy.types.password_hash.base import HashingBackend
+
+if TYPE_CHECKING:
+    from sqlalchemy import BinaryExpression, ColumnElement
+
+
+class PasslibHasher(HashingBackend):
+    """Hashing backend using Passlib.
+
+    Relies on the `passlib` package being installed.
+    Install with `pip install passlib` or `uv pip install passlib`.
+    """
+
+    def __init__(self, context: CryptContext) -> None:
+        """Initialize PasslibBackend.
+
+        Args:
+            context: The Passlib CryptContext to use for hashing and verification.
+        """
+        self.context = context
+
+    def hash(self, value: "Union[str, bytes]") -> str:
+        """Hash the given value using the Passlib context.
+
+        Args:
+            value: The plain text value to hash. Will be converted to string.
+
+        Returns:
+            The hashed string.
+        """
+        return self.context.hash(self._ensure_bytes(value))
+
+    def verify(self, plain: "Union[str, bytes]", hashed: str) -> bool:
+        """Verify a plain text value against a hash using the Passlib context.
+
+        Args:
+            plain: The plain text value to verify. Will be converted to string.
+            hashed: The hash to verify against.
+
+        Returns:
+            True if the plain text matches the hash, False otherwise.
+        """
+        try:
+            return self.context.verify(self._ensure_bytes(plain), hashed)
+        except Exception:  # noqa: BLE001
+            # Passlib can raise various errors for invalid hashes
+            return False
+
+    def compare_expression(self, column: "ColumnElement[str]", plain: Any) -> "BinaryExpression[bool]":
+        """Direct SQL comparison is not supported for Passlib.
+
+        Raises:
+            NotImplementedError: Always raised.
+        """
+        msg = "PasslibHasher does not support direct SQL comparison."
+        raise NotImplementedError(msg)