Radicale/radicale/auth/htpasswd.py

# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2019 Unrud <unrud@outlook.com>
# Copyright © 2024-2025 Peter Bieringer <pb@bieringer.de>
#
# This library is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Radicale.  If not, see <http://www.gnu.org/licenses/>.

"""
Authentication backend that checks credentials with a htpasswd file.

Apache's htpasswd command (httpd.apache.org/docs/programs/htpasswd.html)
manages a file for storing user credentials. It can encrypt passwords using
different the methods BCRYPT/SHA256/SHA512 or MD5-APR1 (a version of MD5 modified for
Apache). MD5-APR1 provides medium security as of 2015. Only BCRYPT/SHA256/SHA512 can be
considered secure by current standards.

MD5-APR1-encrypted credentials can be written by all versions of htpasswd (it
is the default, in fact), whereas BCRYPT/SHA256/SHA512 requires htpasswd 2.4.x or newer.

The `is_authenticated(user, password)` function provided by this module
verifies the user-given credentials by parsing the htpasswd credential file
pointed to by the ``htpasswd_filename`` configuration value while assuming
the password encryption method specified via the ``htpasswd_encryption``
configuration value.

The following htpasswd password encryption methods are supported by Radicale
out-of-the-box:
    - plain-text (created by htpasswd -p ...) -- INSECURE
    - MD5-APR1   (htpasswd -m ...) -- htpasswd's default method, INSECURE
    - SHA256     (htpasswd -2 ...)
    - SHA512     (htpasswd -5 ...)

When bcrypt is installed:
    - BCRYPT     (htpasswd -B ...) -- Requires htpasswd 2.4.x

"""

import functools
import hmac
import os
import threading
import time
from typing import Any, Tuple

from passlib.hash import apr_md5_crypt, sha256_crypt, sha512_crypt

from radicale import auth, config, logger


class Auth(auth.BaseAuth):

    _filename: str
    _encoding: str
    _htpasswd: dict             # login -> digest
    _htpasswd_mtime_ns: int
    _htpasswd_size: int
    _htpasswd_ok: bool
    _htpasswd_not_ok_time: float
    _htpasswd_not_ok_reminder_seconds: int
    _htpasswd_bcrypt_use: int
    _htpasswd_cache: bool
    _has_bcrypt: bool
    _lock: threading.Lock

    def __init__(self, configuration: config.Configuration) -> None:
        super().__init__(configuration)
        self._filename = configuration.get("auth", "htpasswd_filename")
        self._encoding = configuration.get("encoding", "stock")
        self._htpasswd_cache = configuration.get("auth", "htpasswd_cache")
        logger.info("auth htpasswd cache: %s", self._htpasswd_cache)
        encryption: str = configuration.get("auth", "htpasswd_encryption")
        logger.info("auth htpasswd encryption is 'radicale.auth.htpasswd_encryption.%s'", encryption)

        self._has_bcrypt = False
        self._htpasswd_ok = False
        self._htpasswd_not_ok_reminder_seconds = 60 # currently hardcoded
        (self._htpasswd_ok, self._htpasswd_bcrypt_use, self._htpasswd, self._htpasswd_size, self._htpasswd_mtime_ns) = self._read_htpasswd(True, False)
        self._lock = threading.Lock()

        if encryption == "plain":
            self._verify = self._plain
        elif encryption == "md5":
            self._verify = self._md5apr1
        elif encryption == "sha256":
            self._verify = self._sha256
        elif encryption == "sha512":
            self._verify = self._sha512
        elif encryption == "bcrypt" or encryption == "autodetect":
            try:
                import bcrypt
            except ImportError as e:
                if (encryption == "autodetect") and (self._htpasswd_bcrypt_use == 0):
                    logger.warning("auth htpasswd encryption is 'radicale.auth.htpasswd_encryption.%s' which can require bycrypt module, but currently no entries found", encryption)
                else:
                    raise RuntimeError(
                        "The htpasswd encryption method 'bcrypt' or 'autodetect' requires "
                        "the bcrypt module (entries found: %d)." % self._htpasswd_bcrypt_use) from e
            else:
                if encryption == "autodetect":
                    if self._htpasswd_bcrypt_use == 0:
                        logger.info("auth htpasswd encryption is 'radicale.auth.htpasswd_encryption.%s' and bycrypt module found, but currently not required", encryption)
                    else:
                        logger.info("auth htpasswd encryption is 'radicale.auth.htpasswd_encryption.%s' and bycrypt module found (bcrypt entries found: %d)", encryption, self._htpasswd_bcrypt_use)
            if encryption == "bcrypt":
                self._verify = functools.partial(self._bcrypt, bcrypt)
            else:
                self._verify = self._autodetect
                self._verify_bcrypt = functools.partial(self._bcrypt, bcrypt)
            self._has_bcrypt = True
        else:
            raise RuntimeError("The htpasswd encryption method %r is not "
                               "supported." % encryption)

    def _plain(self, hash_value: str, password: str) -> tuple[str, bool]:
        """Check if ``hash_value`` and ``password`` match, plain method."""
        return ("PLAIN", hmac.compare_digest(hash_value.encode(), password.encode()))

    def _bcrypt(self, bcrypt: Any, hash_value: str, password: str) -> tuple[str, bool]:
        return ("BCRYPT", bcrypt.checkpw(password=password.encode('utf-8'), hashed_password=hash_value.encode()))

    def _md5apr1(self, hash_value: str, password: str) -> tuple[str, bool]:
        return ("MD5-APR1", apr_md5_crypt.verify(password, hash_value.strip()))

    def _sha256(self, hash_value: str, password: str) -> tuple[str, bool]:
        return ("SHA-256", sha256_crypt.verify(password, hash_value.strip()))

    def _sha512(self, hash_value: str, password: str) -> tuple[str, bool]:
        return ("SHA-512", sha512_crypt.verify(password, hash_value.strip()))

    def _autodetect(self, hash_value: str, password: str) -> tuple[str, bool]:
        if hash_value.startswith("$apr1$", 0, 6) and len(hash_value) == 37:
            # MD5-APR1
            return self._md5apr1(hash_value, password)
        elif hash_value.startswith("$2y$", 0, 4) and len(hash_value) == 60:
            # BCRYPT
            return self._verify_bcrypt(hash_value, password)
        elif hash_value.startswith("$5$", 0, 3) and len(hash_value) == 63:
            # SHA-256
            return self._sha256(hash_value, password)
        elif hash_value.startswith("$6$", 0, 3) and len(hash_value) == 106:
            # SHA-512
            return self._sha512(hash_value, password)
        else:
            # assumed plaintext
            return self._plain(hash_value, password)

    def _read_htpasswd(self, init: bool, suppress: bool) -> Tuple[bool, int, dict]:
        """Read htpasswd file

        init == True: stop on error
        init == False: warn/skip on error and set mark to log reminder every interval
        suppress == True: suppress warnings, change info to debug (used in non-caching mode)
        suppress == False: do not suppress warnings (used in caching mode)

        """
        htpasswd_ok = True
        bcrypt_use = 0
        if (init is True) or (suppress is True):
            info = "Read"
        else:
            info = "Re-read"
        if suppress is False:
            logger.info("%s content of htpasswd file start: %r", info, self._filename)
        else:
            logger.debug("%s content of htpasswd file start: %r", info, self._filename)
        htpasswd: dict[str, str] = dict()
        entries = 0
        duplicates = 0
        errors = 0
        try:
            with open(self._filename, encoding=self._encoding) as f:
                line_num = 0
                for line in f:
                    line_num += 1
                    line = line.rstrip("\n")
                    if line.lstrip() and not line.lstrip().startswith("#"):
                        try:
                            login, digest = line.split(":", maxsplit=1)
                            skip = False
                            if login == "" or digest == "":
                                if init is True:
                                    raise ValueError("htpasswd file contains problematic line not matching <login>:<digest> in line: %d" % line_num)
                                else:
                                    errors += 1
                                    logger.warning("htpasswd file contains problematic line not matching <login>:<digest> in line: %d (ignored)", line_num)
                                    htpasswd_ok = False
                                    skip = True
                            else:
                                if htpasswd.get(login):
                                    duplicates += 1
                                    if init is True:
                                        raise ValueError("htpasswd file contains duplicate login: '%s'", login, line_num)
                                    else:
                                        logger.warning("htpasswd file contains duplicate login: '%s' (line: %d / ignored)", login, line_num)
                                        htpasswd_ok = False
                                        skip = True
                                else:
                                    if digest.startswith("$2y$", 0, 4) and len(digest) == 60:
                                        if init is True:
                                            bcrypt_use += 1
                                        else:
                                            if self._has_bcrypt is False:
                                                logger.warning("htpasswd file contains bcrypt digest login: '%s' (line: %d / ignored because module is not loaded)", login, line_num)
                                                skip = True
                                                htpasswd_ok = False
                            if skip is False:
                                htpasswd[login] = digest
                                entries += 1
                        except ValueError as e:
                            if init is True:
                                raise RuntimeError("Invalid htpasswd file %r: %s" % (self._filename, e)) from e
        except OSError as e:
            if init is True:
                raise RuntimeError("Failed to load htpasswd file %r: %s" % (self._filename, e)) from e
            else:
                logger.warning("Failed to load htpasswd file on re-read: %r" % self._filename)
                htpasswd_ok = False
        htpasswd_size = os.stat(self._filename).st_size
        htpasswd_mtime_ns = os.stat(self._filename).st_mtime_ns
        if suppress is False:
            logger.info("%s content of htpasswd file done: %r (entries: %d, duplicates: %d, errors: %d)", info, self._filename, entries, duplicates, errors)
        else:
            logger.debug("%s content of htpasswd file done: %r (entries: %d, duplicates: %d, errors: %d)", info, self._filename, entries, duplicates, errors)
        if htpasswd_ok is True:
            self._htpasswd_not_ok_time = 0
        else:
            self._htpasswd_not_ok_time = time.time()
        return (htpasswd_ok, bcrypt_use, htpasswd, htpasswd_size, htpasswd_mtime_ns)

    def _login(self, login: str, password: str) -> str:
        """Validate credentials.

        Iterate through htpasswd credential file until login matches, extract
        hash (encrypted password) and check hash against password,
        using the method specified in the Radicale config.

        The content of the file is cached and live updates will be detected by
        comparing mtime_ns and size

        """
        if self._htpasswd_cache is True:
            # check and re-read file if required
            with self._lock:
                htpasswd_size = os.stat(self._filename).st_size
                htpasswd_mtime_ns = os.stat(self._filename).st_mtime_ns
                if (htpasswd_size != self._htpasswd_size) or (htpasswd_mtime_ns != self._htpasswd_mtime_ns):
                    (self._htpasswd_ok, self._htpasswd_bcrypt_use, self._htpasswd, self._htpasswd_size, self._htpasswd_mtime_ns) = self._read_htpasswd(False, False)
                    self._htpasswd_not_ok_time = 0
        else:
            # read file on every request
            (self._htpasswd_ok, self._htpasswd_bcrypt_use, self._htpasswd, self._htpasswd_size, self._htpasswd_mtime_ns) = self._read_htpasswd(False, True)

        # log reminder of problemantic file every interval
        current_time = time.time()
        if (self._htpasswd_ok is False):
            if (self._htpasswd_not_ok_time > 0):
                if (current_time - self._htpasswd_not_ok_time) > self._htpasswd_not_ok_reminder_seconds:
                    logger.warning("htpasswd file still contains issues (REMINDER, check warnings in the past): %r" % self._filename)
                    self._htpasswd_not_ok_time = current_time
            else:
                self._htpasswd_not_ok_time = current_time

        if self._htpasswd.get(login):
            digest = self._htpasswd[login]
            (method, password_ok) = self._verify(digest, password)
            logger.debug("Login verification successful for user: '%s' (method '%s')", login, method)
            if password_ok:
                return login
            else:
                logger.debug("Login verification failed for user: '%s' ( method '%s')", login, method)
        else:
            logger.debug("Login verification user not found: '%s'", login)
        return ""