pam_ssh_2fa.py

#!/usr/bin/env python3
"""
PAM SSH 2FA Module - Push Notification One-Time Password Authentication
========================================================================

This module provides two-factor authentication for SSH by generating a
random 4-digit code, sending it via push notification (using Apprise),
and prompting the user to enter the code.

OVERVIEW
--------
When a user attempts to authenticate via SSH, this module:
1. Generates a cryptographically secure 4-digit code
2. Sends that code to the user via configured notification service(s)
3. Prompts the user to enter the received code
4. Validates the entered code against the generated one
5. Allows or denies authentication based on the result

INSTALLATION
------------
This module requires:
- libpam-python (provides pam_python.so)
- apprise (Python library for notifications)
- Python 3.6+

Install via the provided install.sh script or manually:
    apt install libpam-python python3-pip
    pip3 install apprise --break-system-packages

CONFIGURATION
-------------
Configuration is stored in /etc/pam-ssh-2fa/config.ini
See config.ini for all available options.

PAM CONFIGURATION
-----------------
Add to /etc/pam.d/sshd (after @include common-auth or as needed):
    auth required pam_python.so /etc/pam-ssh-2fa/pam_ssh_2fa.py

SSH CONFIGURATION
-----------------
In /etc/ssh/sshd_config:
    UsePAM yes
    KbdInteractiveAuthentication yes
    AuthenticationMethods publickey,keyboard-interactive:pam

DEBUGGING
---------
- Set debug = true in config.ini to enable verbose logging
- Logs go to the configured log_file (default: /var/log/pam-ssh-2fa.log)
- Also logs to syslog under the AUTH facility
- Test with: pamtester sshd <username> authenticate

SECURITY CONSIDERATIONS
-----------------------
- Codes expire after a configurable timeout (default: 5 minutes)
- Codes are single-use and invalidated after successful authentication
- Failed attempts are logged with source IP
- Code storage uses secure file permissions (0600)
- Consider rate limiting at the firewall level for brute force protection

AUTHOR
------
Generated by Claude (Anthropic) for SSH 2FA push notification authentication.

LICENSE
-------
MIT License - Use and modify freely.

VERSION
-------
1.0.0 - Initial release
"""

# =============================================================================
# IMPORTS
# =============================================================================

import os
import sys
import time
import secrets
import configparser
import logging
import json
import hashlib
from pathlib import Path
from datetime import datetime, timedelta
from typing import Optional, Tuple, Dict, Any

# Apprise is imported conditionally to provide better error messages
# if it's not installed
try:
    import apprise

    APPRISE_AVAILABLE = True
except ImportError:
    APPRISE_AVAILABLE = False


# =============================================================================
# CONSTANTS
# =============================================================================

# Default configuration file location
CONFIG_FILE = "/etc/pam-ssh-2fa/config.ini"

# Default values if config file is missing or incomplete
DEFAULTS = {
    "code_length": 4,  # Length of the OTP code
    "code_timeout": 300,  # Seconds before code expires (5 minutes)
    "max_attempts": 3,  # Maximum entry attempts before failure
    "debug": False,  # Enable debug logging
    "log_file": "/var/log/pam-ssh-2fa.log",
    "code_storage_dir": "/var/run/pam-ssh-2fa",
    "apprise_urls": "",  # Comma-separated Apprise notification URLs
    "notification_title": "SSH Login Code",
    "notification_body": "Your SSH verification code is: {code}\n\nHost: {host}\nUser: {user}\nFrom: {rhost}\n\nThis code expires in {timeout} minutes.",
    "notification_body_link": "Click to approve SSH login:\n{link}\n\nHost: {host}\nUser: {user}\nFrom: {rhost}\n\nThis link expires in {timeout} minutes.",
    "notification_body_both": "SSH Login for {user}@{host}\n\nClick to approve:\n{link}\n\nOr enter code: {code}\n\nFrom: {rhost}\nExpires in {timeout} minutes.",
    "prompt_message": "Enter verification code: ",
    "prompt_message_both": "Enter code OR press Enter after clicking link: ",
    "success_message": "Verification successful.",
    "failure_message": "Verification failed.",
    "expired_message": "Code expired. Please reconnect to receive a new code.",
    "bypass_users": "",  # Comma-separated list of users to skip 2FA
    "bypass_networks": "",  # Comma-separated CIDR ranges to skip 2FA
    "allow_unconfigured_users": False,  # If True, users without config bypass 2FA
    "auth_method": "code",  # Default auth method: code, link, both, none
    "server_port": 9110,  # Approval server port
    "server_url": "",  # Approval server URL (e.g., http://myserver.com:9110)
}

# PAM return codes (defined here for clarity)
PAM_SUCCESS = 0
PAM_AUTH_ERR = 7
PAM_CRED_INSUFFICIENT = 8
PAM_AUTHINFO_UNAVAIL = 9
PAM_USER_UNKNOWN = 10
PAM_MAXTRIES = 11
PAM_IGNORE = 25


# =============================================================================
# LOGGING SETUP
# =============================================================================


class PAMLogger:
    """
    Custom logger for PAM module that writes to both file and syslog.

    This class provides a unified logging interface that:
    - Writes to a dedicated log file for easy debugging
    - Also logs to syslog for system integration
    - Supports debug mode for verbose output
    - Sanitizes sensitive data before logging

    Attributes:
        log_file (str): Path to the log file
        debug (bool): Whether debug logging is enabled
        logger (logging.Logger): The underlying Python logger instance

    Usage:
        logger = PAMLogger("/var/log/pam-ssh-2fa.log", debug=True)
        logger.info("User authenticated", user="john", rhost="192.168.1.1")
        logger.debug("Code generated", code_hash="abc123")  # Only if debug=True
    """

    def __init__(self, log_file: str, debug: bool = False):
        """
        Initialize the PAM logger.

        Args:
            log_file: Path to the log file. Directory will be created if needed.
            debug: If True, DEBUG level messages will be logged.
        """
        self.log_file = log_file
        self.debug = debug
        self.logger = logging.getLogger("pam-ssh-2fa")
        self.logger.setLevel(logging.DEBUG if debug else logging.INFO)

        # Clear any existing handlers to avoid duplicates on module reload
        self.logger.handlers.clear()

        # Create log directory if it doesn't exist
        log_dir = os.path.dirname(log_file)
        if log_dir and not os.path.exists(log_dir):
            try:
                os.makedirs(log_dir, mode=0o750)
            except OSError:
                pass  # Fall back to syslog only

        # File handler with detailed format
        try:
            file_handler = logging.FileHandler(log_file)
            file_handler.setFormatter(
                logging.Formatter(
                    "%(asctime)s [%(levelname)s] %(message)s",
                    datefmt="%Y-%m-%d %H:%M:%S",
                )
            )
            self.logger.addHandler(file_handler)
        except (OSError, IOError):
            pass  # Fall back to syslog only

        # Syslog handler for system integration
        try:
            from logging.handlers import SysLogHandler

            syslog_handler = SysLogHandler(
                address="/dev/log", facility=SysLogHandler.LOG_AUTH
            )
            syslog_handler.setFormatter(logging.Formatter("pam-ssh-2fa: %(message)s"))
            self.logger.addHandler(syslog_handler)
        except (OSError, IOError):
            pass  # Syslog not available

    def _format_message(self, message: str, **kwargs) -> str:
        """
        Format a log message with additional context.

        Args:
            message: The main log message
            **kwargs: Additional key-value pairs to include

        Returns:
            Formatted message string with context appended
        """
        if kwargs:
            context = " | ".join(f"{k}={v}" for k, v in kwargs.items())
            return f"{message} | {context}"
        return message

    def info(self, message: str, **kwargs):
        """Log an INFO level message."""
        self.logger.info(self._format_message(message, **kwargs))

    def warning(self, message: str, **kwargs):
        """Log a WARNING level message."""
        self.logger.warning(self._format_message(message, **kwargs))

    def error(self, message: str, **kwargs):
        """Log an ERROR level message."""
        self.logger.error(self._format_message(message, **kwargs))

    def debug(self, message: str, **kwargs):
        """Log a DEBUG level message (only if debug mode is enabled)."""
        if self.debug:
            self.logger.debug(self._format_message(message, **kwargs))


# =============================================================================
# CONFIGURATION MANAGEMENT
# =============================================================================


class Config:
    """
    Configuration manager for the PAM 2FA module.

    Loads configuration from an INI file and provides access to settings
    with sensible defaults. Configuration is validated on load.

    Supports per-user configuration files for notification settings,
    allowing different users to receive codes via different services.

    Configuration hierarchy (highest priority first):
    1. Per-user config: /etc/pam-ssh-2fa/users/<username>.conf
    2. Global config: /etc/pam-ssh-2fa/config.ini
    3. Built-in defaults

    The configuration file uses standard INI format:

        [general]
        debug = false
        log_file = /var/log/pam-ssh-2fa.log

        [codes]
        length = 4
        timeout = 300

        [notifications]
        apprise_urls = ntfy://ntfy.sh/my-topic,pover://...

    Per-user config files only need the [notifications] section:

        # /etc/pam-ssh-2fa/users/doug.conf
        [notifications]
        apprise_urls = pover://DOUG_USER_KEY@APP_TOKEN

    Attributes:
        config_file (str): Path to the main configuration file
        user_config_dir (str): Path to per-user config directory
        settings (dict): Parsed configuration settings

    Usage:
        config = Config("/etc/pam-ssh-2fa/config.ini")
        timeout = config.get("code_timeout", 300)
        urls = config.get_list("apprise_urls")

        # Get user-specific config
        user_config = config.get_user_config("doug")
        doug_urls = user_config.get_list("apprise_urls")
    """

    def __init__(self, config_file: str = CONFIG_FILE, user: str = None):
        """
        Initialize configuration from file.

        Args:
            config_file: Path to the INI configuration file.
                        If file doesn't exist, defaults are used.
            user: Optional username to load user-specific overrides.
        """
        self.config_file = config_file
        self.user = user
        self.settings = DEFAULTS.copy()

        # Derive user config directory from main config file location
        config_dir = os.path.dirname(config_file) or "/etc/pam-ssh-2fa"
        self.user_config_dir = os.path.join(config_dir, "users")

        # Load global config first
        self._load_config()

        # Then overlay user-specific config if user provided
        if user:
            self._load_user_config(user)

    def _load_config(self):
        """
        Load and parse the main configuration file.

        Reads the INI file and maps sections/keys to flat settings dict.
        Missing values use defaults. Invalid values are logged and skipped.
        """
        if not os.path.exists(self.config_file):
            return  # Use defaults

        self._parse_config_file(self.config_file)

    def _load_user_config(self, user: str):
        """
        Load user-specific configuration overrides.

        Looks for a config file named after the user in the users/ directory.
        User config only overrides notification settings, not system settings.

        Args:
            user: Username to load config for
        """
        # Sanitize username to prevent directory traversal
        safe_user = "".join(c for c in user if c.isalnum() or c in "-_.")
        if not safe_user or safe_user != user:
            return  # Invalid username, skip user config

        # Try multiple file extensions
        for ext in [".conf", ".ini", ""]:
            user_config_file = os.path.join(self.user_config_dir, f"{safe_user}{ext}")
            if os.path.exists(user_config_file):
                self._parse_config_file(user_config_file, user_only=True)
                return

    def _parse_config_file(self, filepath: str, user_only: bool = False):
        """
        Parse a configuration file and update settings.

        Args:
            filepath: Path to the config file
            user_only: If True, only load user-customizable settings
                      (notifications section). This prevents users from
                      overriding system-wide security settings.
        """
        parser = configparser.ConfigParser()
        try:
            parser.read(filepath)
        except configparser.Error:
            return  # Skip on parse error

        # Full mapping for system config
        section_mapping = {
            ("general", "debug"): ("debug", self._parse_bool),
            ("general", "log_file"): ("log_file", str),
            ("codes", "length"): ("code_length", int),
            ("codes", "timeout"): ("code_timeout", int),
            ("codes", "max_attempts"): ("max_attempts", int),
            ("codes", "storage_dir"): ("code_storage_dir", str),
            ("notifications", "apprise_urls"): ("apprise_urls", str),
            ("notifications", "title"): ("notification_title", str),
            ("notifications", "body"): ("notification_body", str),
            ("notifications", "body_link"): ("notification_body_link", str),
            ("notifications", "body_both"): ("notification_body_both", str),
            ("messages", "prompt"): ("prompt_message", str),
            ("messages", "prompt_both"): ("prompt_message_both", str),
            ("messages", "success"): ("success_message", str),
            ("messages", "failure"): ("failure_message", str),
            ("messages", "expired"): ("expired_message", str),
            ("bypass", "users"): ("bypass_users", str),
            ("bypass", "networks"): ("bypass_networks", str),
            ("users", "allow_unconfigured_users"): (
                "allow_unconfigured_users",
                self._parse_bool,
            ),
            ("users", "auth_method"): ("auth_method", str),
            ("server", "port"): ("server_port", int),
            ("server", "url"): ("server_url", str),
        }

        # User configs can only override notification settings
        # This prevents privilege escalation via user config files
        if user_only:
            section_mapping = {
                ("notifications", "apprise_urls"): ("apprise_urls", str),
                ("notifications", "title"): ("notification_title", str),
                ("notifications", "body"): ("notification_body", str),
                ("notifications", "body_link"): ("notification_body_link", str),
                ("notifications", "body_both"): ("notification_body_both", str),
                ("auth", "method"): ("auth_method", str),
            }

        for (section, key), (setting_name, converter) in section_mapping.items():
            if parser.has_option(section, key):
                try:
                    raw_value = parser.get(section, key)
                    self.settings[setting_name] = converter(raw_value)
                except (ValueError, TypeError):
                    pass  # Keep default on conversion error

    def _parse_bool(self, value: str) -> bool:
        """
        Parse a string to boolean.

        Args:
            value: String like "true", "yes", "1", "false", "no", "0"

        Returns:
            Boolean interpretation of the string
        """
        return value.lower() in ("true", "yes", "1", "on", "enabled")

    def get(self, key: str, default: Any = None) -> Any:
        """
        Get a configuration value.

        Args:
            key: The setting name
            default: Value to return if key not found

        Returns:
            The configuration value or default
        """
        return self.settings.get(key, default)

    def get_list(self, key: str, separator: str = ",") -> list:
        """
        Get a configuration value as a list.

        Splits the string value by separator and strips whitespace.

        Args:
            key: The setting name
            separator: Character to split on (default: comma)

        Returns:
            List of strings, empty list if key not found or empty
        """
        value = self.settings.get(key, "")
        if not value:
            return []
        return [item.strip() for item in value.split(separator) if item.strip()]

    def get_user_config(self, user: str) -> "Config":
        """
        Get a new Config instance with user-specific overrides applied.

        This is useful when you need to get settings for a specific user
        after already loading the global config.

        Args:
            user: Username to load config for

        Returns:
            New Config instance with user overrides applied
        """
        return Config(self.config_file, user=user)

    def has_user_config(self, user: str) -> bool:
        """
        Check if a user-specific configuration file exists.

        Args:
            user: Username to check

        Returns:
            True if user has a config file
        """
        safe_user = "".join(c for c in user if c.isalnum() or c in "-_.")
        if not safe_user or safe_user != user:
            return False

        for ext in [".conf", ".ini", ""]:
            user_config_file = os.path.join(self.user_config_dir, f"{safe_user}{ext}")
            if os.path.exists(user_config_file):
                return True
        return False


# =============================================================================
# CODE GENERATION AND STORAGE
# =============================================================================


class CodeManager:
    """
    Manages generation, storage, and validation of one-time codes.

    Codes are stored in individual files named by a hash of the username
    and session info. This allows multiple pending codes for different
    sessions while preventing enumeration of usernames.

    File format is JSON with:
    - code: The plaintext OTP code
    - created: Unix timestamp of creation
    - expires: Unix timestamp of expiration
    - user: Username for verification
    - rhost: Remote host IP
    - attempts: Number of validation attempts made

    Security features:
    - Codes stored with 0600 permissions
    - Storage directory has 0700 permissions
    - Codes automatically expire
    - Codes are deleted after successful use
    - Failed attempts are tracked and limited

    Attributes:
        storage_dir (str): Directory for code storage files
        code_length (int): Length of generated codes
        timeout (int): Seconds until code expiration
        max_attempts (int): Maximum validation attempts allowed
        logger (PAMLogger): Logger instance for debugging

    Usage:
        manager = CodeManager("/var/run/pam-ssh-2fa", logger=logger)
        code = manager.generate("john", "192.168.1.1")
        is_valid = manager.validate("john", "192.168.1.1", "1234")
    """

    def __init__(
        self,
        storage_dir: str,
        code_length: int = 4,
        timeout: int = 300,
        max_attempts: int = 3,
        logger: Optional[PAMLogger] = None,
    ):
        """
        Initialize the code manager.

        Args:
            storage_dir: Directory to store code files
            code_length: Number of digits in generated codes
            timeout: Seconds until codes expire
            max_attempts: Max validation attempts before lockout
            logger: Optional logger for debug output
        """
        self.storage_dir = storage_dir
        self.code_length = code_length
        self.timeout = timeout
        self.max_attempts = max_attempts
        self.logger = logger

        # Ensure storage directory exists with secure permissions
        self._ensure_storage_dir()

    def _ensure_storage_dir(self):
        """
        Create the storage directory if it doesn't exist.

        Sets permissions to 0700 (owner read/write/execute only).
        This prevents other users from accessing pending codes.
        """
        if not os.path.exists(self.storage_dir):
            try:
                os.makedirs(self.storage_dir, mode=0o700)
            except OSError as e:
                if self.logger:
                    self.logger.error(f"Failed to create storage dir: {e}")
                raise
        else:
            # Ensure correct permissions even if dir already exists
            try:
                os.chmod(self.storage_dir, 0o700)
            except OSError:
                pass

    def _get_code_file(self, user: str, rhost: str) -> str:
        """
        Get the path to the code file for a user/session.

        Uses a hash of user and rhost to prevent username enumeration
        while still allowing session-specific codes.

        Args:
            user: Username
            rhost: Remote host IP address

        Returns:
            Full path to the code file
        """
        # Create a hash of user+rhost for the filename
        # This prevents enumeration of usernames from the filesystem
        session_key = f"{user}:{rhost}"
        file_hash = hashlib.sha256(session_key.encode()).hexdigest()[:16]
        return os.path.join(self.storage_dir, f"code_{file_hash}.json")

    def generate(self, user: str, rhost: str) -> str:
        """
        Generate a new one-time code for a user session.

        Creates a cryptographically secure random code and stores it
        with metadata for later validation.

        Args:
            user: Username requesting authentication
            rhost: Remote host IP address

        Returns:
            The generated code as a string (e.g., "1234")

        Raises:
            OSError: If unable to write the code file
        """
        # Generate cryptographically secure random code
        # Using secrets module for security-sensitive randomness
        code = "".join(str(secrets.randbelow(10)) for _ in range(self.code_length))

        # Calculate expiration time
        now = time.time()
        expires = now + self.timeout

        # Prepare code data
        code_data = {
            "code": code,
            "created": now,
            "expires": expires,
            "user": user,
            "rhost": rhost,
            "attempts": 0,
        }

        # Write to file with secure permissions
        code_file = self._get_code_file(user, rhost)

        # Create file with restricted permissions from the start
        # Using os.open with mode flags to ensure atomic secure creation
        fd = os.open(code_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            with os.fdopen(fd, "w") as f:
                json.dump(code_data, f)
        except Exception:
            os.close(fd)
            raise

        if self.logger:
            # Log code generation (hash the code for security in logs)
            code_hash = hashlib.sha256(code.encode()).hexdigest()[:8]
            self.logger.debug(
                "Code generated",
                user=user,
                rhost=rhost,
                code_hash=code_hash,
                expires_in=f"{self.timeout}s",
            )

        return code

    def validate(self, user: str, rhost: str, entered_code: str) -> Tuple[bool, str]:
        """
        Validate an entered code against the stored code.

        Checks if:
        - A code exists for this user/session
        - The code hasn't expired
        - Maximum attempts haven't been exceeded
        - The entered code matches

        On successful validation, the code file is deleted.
        On failure, the attempt counter is incremented.

        Args:
            user: Username
            rhost: Remote host IP address
            entered_code: The code entered by the user

        Returns:
            Tuple of (success: bool, message: str)
            - success: True if code was valid
            - message: Human-readable result message
        """
        code_file = self._get_code_file(user, rhost)

        # Check if code file exists
        if not os.path.exists(code_file):
            if self.logger:
                self.logger.warning(
                    "No code found for validation", user=user, rhost=rhost
                )
            return False, "No pending code found. Please reconnect."

        # Read and parse code data
        try:
            with open(code_file, "r") as f:
                code_data = json.load(f)
        except (json.JSONDecodeError, IOError) as e:
            if self.logger:
                self.logger.error(f"Failed to read code file: {e}")
            self._cleanup_code(code_file)
            return False, "Internal error. Please reconnect."

        # Verify the code is for this user (defense in depth)
        if code_data.get("user") != user:
            if self.logger:
                self.logger.warning(
                    "User mismatch in code validation",
                    expected=code_data.get("user"),
                    got=user,
                )
            return False, "Session mismatch. Please reconnect."

        # Check if code has expired
        if time.time() > code_data.get("expires", 0):
            if self.logger:
                self.logger.info("Code expired", user=user, rhost=rhost)
            self._cleanup_code(code_file)
            return False, "Code expired. Please reconnect to receive a new code."

        # Check attempt count
        attempts = code_data.get("attempts", 0)
        if attempts >= self.max_attempts:
            if self.logger:
                self.logger.warning(
                    "Max attempts exceeded", user=user, rhost=rhost, attempts=attempts
                )
            self._cleanup_code(code_file)
            return False, "Maximum attempts exceeded. Please reconnect."

        # Validate the code using constant-time comparison
        # This prevents timing attacks that could reveal code characters
        stored_code = code_data.get("code", "")
        if secrets.compare_digest(entered_code.strip(), stored_code):
            # Success! Clean up the code file
            if self.logger:
                self.logger.info("Code validated successfully", user=user, rhost=rhost)
            self._cleanup_code(code_file)
            return True, "Verification successful."

        # Invalid code - increment attempt counter
        code_data["attempts"] = attempts + 1
        remaining = self.max_attempts - code_data["attempts"]

        try:
            with open(code_file, "w") as f:
                json.dump(code_data, f)
        except IOError:
            pass  # Best effort to save attempts

        if self.logger:
            self.logger.warning(
                "Invalid code entered",
                user=user,
                rhost=rhost,
                attempts=code_data["attempts"],
                remaining=remaining,
            )

        if remaining > 0:
            return False, f"Invalid code. {remaining} attempts remaining."
        else:
            self._cleanup_code(code_file)
            return False, "Invalid code. Maximum attempts exceeded."

    def _cleanup_code(self, code_file: str):
        """
        Securely remove a code file.

        Args:
            code_file: Path to the code file to remove
        """
        try:
            os.unlink(code_file)
        except OSError:
            pass  # Best effort cleanup

    def cleanup_expired(self):
        """
        Remove all expired code files from storage.

        This can be called periodically (e.g., via cron) to clean up
        abandoned codes from interrupted sessions.
        """
        if not os.path.exists(self.storage_dir):
            return

        now = time.time()
        for filename in os.listdir(self.storage_dir):
            if not filename.startswith("code_"):
                continue

            filepath = os.path.join(self.storage_dir, filename)
            try:
                with open(filepath, "r") as f:
                    code_data = json.load(f)

                if now > code_data.get("expires", 0):
                    os.unlink(filepath)
                    if self.logger:
                        self.logger.debug(f"Cleaned up expired code: {filename}")
            except (json.JSONDecodeError, IOError, OSError):
                # Remove corrupt files
                try:
                    os.unlink(filepath)
                except OSError:
                    pass


# =============================================================================
# APPROVAL MANAGER (for link-based authentication)
# =============================================================================


class ApprovalManager:
    """
    Manages link-based authentication approvals.

    When auth_method is 'link' or 'both', this manager creates approval
    request files that the approval server monitors. When a user clicks
    the approval link, the server marks the file as approved.

    Approval files are JSON with structure:
    {
        "token": "abc123...",
        "user": "username",
        "rhost": "192.168.1.1",
        "host": "servername",
        "created": 1234567890.123,
        "expires": 1234567890.123,
        "approved": false,
        "approved_at": null
    }

    Attributes:
        approvals_dir (str): Directory for approval request files
        timeout (int): Seconds until approval expires
        server_url (str): Base URL for approval links
        logger (PAMLogger): Optional logger for debugging

    Usage:
        manager = ApprovalManager(
            approvals_dir="/var/run/pam-ssh-2fa/approvals",
            timeout=300,
            server_url="http://myserver.com:9110",
            logger=logger
        )
        token, link = manager.create_approval("john", "192.168.1.1")
        # Send link to user...
        # Poll for approval:
        while not manager.is_approved(token):
            time.sleep(1)
    """

    def __init__(
        self,
        approvals_dir: str,
        timeout: int = 300,
        server_url: str = "",
        logger: Optional[PAMLogger] = None,
    ):
        """
        Initialize the approval manager.

        Args:
            approvals_dir: Directory to store approval files
            timeout: Seconds until approvals expire
            server_url: Base URL for approval server (e.g., http://myserver:9110)
            logger: Optional logger for debug output
        """
        self.approvals_dir = approvals_dir
        self.timeout = timeout
        self.server_url = server_url.rstrip("/") if server_url else ""
        self.logger = logger

        # Ensure approvals directory exists
        self._ensure_dir()

    def _ensure_dir(self):
        """Create the approvals directory if it doesn't exist."""
        if not os.path.exists(self.approvals_dir):
            try:
                os.makedirs(self.approvals_dir, mode=0o700)
            except OSError as e:
                if self.logger:
                    self.logger.error(f"Failed to create approvals dir: {e}")
                raise
        else:
            # Ensure correct permissions
            try:
                os.chmod(self.approvals_dir, 0o700)
            except OSError:
                pass

    def _get_approval_file(self, token: str) -> str:
        """Get the path to an approval file by token."""
        # Sanitize token to prevent directory traversal
        safe_token = "".join(c for c in token if c.isalnum() or c in '-_')
        return os.path.join(self.approvals_dir, f"{safe_token}.json")

    def create_approval(self, user: str, rhost: str) -> Tuple[str, str]:
        """
        Create a new approval request.

        Args:
            user: Username requesting authentication
            rhost: Remote host IP address

        Returns:
            Tuple of (token, approval_link)

        Raises:
            OSError: If unable to write the approval file
            ValueError: If server_url is not configured
        """
        if not self.server_url:
            raise ValueError("server_url not configured - cannot create approval links")

        # Generate cryptographically secure random token
        token = secrets.token_urlsafe(32)

        # Get hostname
        try:
            hostname = os.uname().nodename
        except (AttributeError, OSError):
            hostname = "unknown"

        # Calculate expiration
        now = time.time()
        expires = now + self.timeout

        # Prepare approval data
        approval_data = {
            "token": token,
            "user": user,
            "rhost": rhost,
            "host": hostname,
            "created": now,
            "expires": expires,
            "approved": False,
            "approved_at": None,
        }

        # Write to file with secure permissions
        approval_file = self._get_approval_file(token)

        fd = os.open(approval_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            with os.fdopen(fd, "w") as f:
                json.dump(approval_data, f)
        except Exception:
            os.close(fd)
            raise

        # Construct approval link
        approval_link = f"{self.server_url}/approve/{token}"

        if self.logger:
            self.logger.debug(
                "Approval request created",
                user=user,
                rhost=rhost,
                token=token[:8] + "...",
                expires_in=f"{self.timeout}s",
            )

        return token, approval_link

    def is_approved(self, token: str) -> bool:
        """
        Check if an approval has been granted.

        Args:
            token: The approval token

        Returns:
            True if approved, False otherwise
        """
        approval_file = self._get_approval_file(token)

        if not os.path.exists(approval_file):
            return False

        try:
            with open(approval_file, "r") as f:
                data = json.load(f)

            # Check if expired
            if time.time() > data.get("expires", 0):
                return False

            return data.get("approved", False)

        except (json.JSONDecodeError, IOError):
            return False

    def is_expired(self, token: str) -> bool:
        """
        Check if an approval request has expired.

        Args:
            token: The approval token

        Returns:
            True if expired or not found, False if still valid
        """
        approval_file = self._get_approval_file(token)

        if not os.path.exists(approval_file):
            return True

        try:
            with open(approval_file, "r") as f:
                data = json.load(f)

            return time.time() > data.get("expires", 0)

        except (json.JSONDecodeError, IOError):
            return True

    def cleanup(self, token: str):
        """
        Remove an approval request file.

        Args:
            token: The approval token to clean up
        """
        approval_file = self._get_approval_file(token)
        try:
            os.unlink(approval_file)
        except OSError:
            pass

    def wait_for_approval(
        self, token: str, poll_interval: float = 1.0, timeout: Optional[int] = None
    ) -> bool:
        """
        Wait for an approval to be granted.

        Polls the approval file until approved or timeout.

        Args:
            token: The approval token
            poll_interval: Seconds between polls (default 1.0)
            timeout: Max seconds to wait (default: use approval timeout)

        Returns:
            True if approved, False if timeout/expired
        """
        if timeout is None:
            timeout = self.timeout

        start_time = time.time()

        while True:
            # Check if approved
            if self.is_approved(token):
                return True

            # Check if expired
            if self.is_expired(token):
                return False

            # Check overall timeout
            if time.time() - start_time > timeout:
                return False

            # Wait before next poll
            time.sleep(poll_interval)


# =============================================================================
# NOTIFICATION SENDER
# =============================================================================


class NotificationSender:
    """
    Sends push notifications via Apprise.

    Apprise supports many notification services including:
    - ntfy (ntfy://)
    - Pushover (pover://)
    - Telegram (tgram://)
    - Email (mailto://)
    - Slack (slack://)
    - Discord (discord://)
    - And many more: https://github.com/caronc/apprise/wiki

    Multiple notification URLs can be configured to send to several
    services simultaneously (redundancy).

    Supports three notification body templates:
    - body_template: For code-only authentication
    - body_template_link: For link-only authentication
    - body_template_both: For combined code + link authentication

    Attributes:
        apprise_urls (list): List of Apprise notification URLs
        title_template (str): Template for notification title
        body_template (str): Template for code-only body
        body_template_link (str): Template for link-only body
        body_template_both (str): Template for code + link body
        logger (PAMLogger): Optional logger for debugging

    Template variables:
        {code} - The OTP code
        {link} - The approval link
        {user} - Username
        {host} - Server hostname
        {rhost} - Remote host IP
        {timeout} - Code timeout in minutes

    Usage:
        sender = NotificationSender(
            ["ntfy://ntfy.sh/my-topic"],
            logger=logger
        )
        success = sender.send(
            code="1234",
            user="john",
            rhost="192.168.1.1",
            timeout=300,
            link="http://server:9110/approve/abc123",
            auth_method="both"
        )
    """

    def __init__(
        self,
        apprise_urls: list,
        title_template: str = "SSH Login Code",
        body_template: str = "Your code is: {code}",
        body_template_link: str = "Click to approve: {link}",
        body_template_both: str = "Click: {link}\n\nOr enter code: {code}",
        logger: Optional[PAMLogger] = None,
    ):
        """
        Initialize the notification sender.

        Args:
            apprise_urls: List of Apprise notification URLs
            title_template: Template string for notification title
            body_template: Template string for code-only body
            body_template_link: Template string for link-only body
            body_template_both: Template string for code + link body
            logger: Optional logger for debugging
        """
        self.apprise_urls = apprise_urls
        self.title_template = title_template
        self.body_template = body_template
        self.body_template_link = body_template_link
        self.body_template_both = body_template_both
        self.logger = logger

        # Validate Apprise is available
        if not APPRISE_AVAILABLE:
            if self.logger:
                self.logger.error(
                    "Apprise module not installed. "
                    "Install with: pip3 install apprise"
                )

    def send(
        self,
        code: str,
        user: str,
        rhost: str,
        timeout: int,
        link: str = None,
        auth_method: str = "code",
    ) -> bool:
        """
        Send a notification with authentication info.

        Attempts to send to all configured notification URLs.
        Returns True if at least one notification succeeds.

        Args:
            code: The OTP code to send (can be empty for link-only)
            user: Username for template
            rhost: Remote host IP for template
            timeout: Code timeout in seconds (converted to minutes for display)
            link: Optional approval link for link-based auth
            auth_method: One of "code", "link", or "both"

        Returns:
            True if at least one notification was sent successfully
        """
        if not APPRISE_AVAILABLE:
            if self.logger:
                self.logger.error("Cannot send notification: Apprise not available")
            return False

        if not self.apprise_urls:
            if self.logger:
                self.logger.error("No notification URLs configured")
            return False

        # Get hostname for template
        try:
            hostname = os.uname().nodename
        except (AttributeError, OSError):
            hostname = "unknown"

        # Prepare template variables
        template_vars = {
            "code": code or "",
            "user": user,
            "host": hostname,
            "rhost": rhost,
            "timeout": str(timeout // 60),  # Convert to minutes
            "link": link or "",
        }

        # Select appropriate body template based on auth method
        if auth_method == "link" and self.body_template_link:
            body_template = self.body_template_link
        elif auth_method == "both" and self.body_template_both:
            body_template = self.body_template_both
        else:
            body_template = self.body_template

        # Format title and body
        try:
            title = self.title_template.format(**template_vars)
            body = body_template.format(**template_vars)
        except KeyError as e:
            if self.logger:
                self.logger.error(f"Template error: unknown variable {e}")
            # Fall back to simple format based on auth method
            title = "SSH Login"
            if auth_method == "link":
                body = f"Click to approve: {link}"
            elif auth_method == "both":
                body = f"Click to approve: {link}\n\nOr enter code: {code}"
            else:
                body = f"Your verification code is: {code}"

        # Create Apprise instance and add notification URLs
        apobj = apprise.Apprise()

        for url in self.apprise_urls:
            apobj.add(url)

        if self.logger:
            self.logger.debug(
                "Sending notification",
                user=user,
                rhost=rhost,
                auth_method=auth_method,
                url_count=len(self.apprise_urls),
            )

        # Send the notification
        # notify() returns True if at least one notification succeeded
        try:
            result = apobj.notify(
                title=title, body=body, notify_type=apprise.NotifyType.INFO
            )

            if self.logger:
                if result:
                    self.logger.info(
                        "Notification sent successfully", user=user, rhost=rhost
                    )
                else:
                    self.logger.error(
                        "All notification attempts failed", user=user, rhost=rhost
                    )

            return result

        except Exception as e:
            if self.logger:
                self.logger.error(f"Notification error: {e}")
            return False


# =============================================================================
# BYPASS CHECKER
# =============================================================================


class BypassChecker:
    """
    Determines if 2FA should be bypassed for certain users or networks.

    Useful for:
    - Service accounts that can't do interactive auth
    - Trusted internal networks
    - Emergency access accounts

    Attributes:
        bypass_users (list): Usernames that skip 2FA
        bypass_networks (list): CIDR ranges that skip 2FA
        logger (PAMLogger): Optional logger for debugging

    Usage:
        checker = BypassChecker(
            bypass_users=["root", "ansible"],
            bypass_networks=["192.168.1.0/24", "10.0.0.0/8"]
        )
        if checker.should_bypass("ansible", "192.168.1.50"):
            return PAM_SUCCESS  # Skip 2FA
    """

    def __init__(
        self,
        bypass_users: list = None,
        bypass_networks: list = None,
        logger: Optional[PAMLogger] = None,
    ):
        """
        Initialize the bypass checker.

        Args:
            bypass_users: List of usernames to bypass
            bypass_networks: List of CIDR ranges to bypass
            logger: Optional logger for debugging
        """
        self.bypass_users = bypass_users or []
        self.bypass_networks = bypass_networks or []
        self.logger = logger

    def should_bypass(self, user: str, rhost: str) -> bool:
        """
        Check if 2FA should be bypassed for this user/connection.

        Args:
            user: Username
            rhost: Remote host IP address

        Returns:
            True if 2FA should be skipped
        """
        # Check user bypass list
        if user in self.bypass_users:
            if self.logger:
                self.logger.info(
                    "2FA bypassed for user", user=user, reason="user_bypass_list"
                )
            return True

        # Check network bypass list
        if self._ip_in_networks(rhost):
            if self.logger:
                self.logger.info(
                    "2FA bypassed for network",
                    user=user,
                    rhost=rhost,
                    reason="network_bypass_list",
                )
            return True

        return False

    def _ip_in_networks(self, ip: str) -> bool:
        """
        Check if an IP address is in any of the bypass networks.

        Uses Python's ipaddress module for proper CIDR matching.

        Args:
            ip: IP address string to check

        Returns:
            True if IP is in any bypass network
        """
        if not self.bypass_networks or not ip:
            return False

        try:
            import ipaddress

            # Parse the remote IP
            try:
                check_ip = ipaddress.ip_address(ip)
            except ValueError:
                return False  # Invalid IP format

            # Check against each bypass network
            for network_str in self.bypass_networks:
                try:
                    network = ipaddress.ip_network(network_str, strict=False)
                    if check_ip in network:
                        return True
                except ValueError:
                    continue  # Invalid network, skip

            return False

        except ImportError:
            # ipaddress module not available (shouldn't happen in Python 3)
            return False


# =============================================================================
# PAM CONVERSATION HELPER
# =============================================================================


def pam_prompt(pamh, message: str, echo: bool = True) -> Optional[str]:
    """
    Prompt the user for input via PAM conversation.

    This function uses PAM's conversation mechanism to display a prompt
    to the user and collect their response. Works with SSH's
    keyboard-interactive authentication.

    Args:
        pamh: PAM handle (provided by pam_python)
        message: The prompt message to display
        echo: If True, show typed characters. If False, hide them (like password)

    Returns:
        The user's response string, or None on error
    """
    # Determine message style based on echo setting
    # PAM_PROMPT_ECHO_ON shows input, PAM_PROMPT_ECHO_OFF hides it
    if echo:
        msg_style = pamh.PAM_PROMPT_ECHO_ON
    else:
        msg_style = pamh.PAM_PROMPT_ECHO_OFF

    try:
        # Create PAM message
        pam_message = pamh.Message(msg_style, message)

        # Send to user and get response
        response = pamh.conversation(pam_message)

        # Response is a Response object, get the actual text
        return response.resp

    except pamh.exception as e:
        # PAM conversation failed (user cancelled, timeout, etc.)
        return None


def pam_info(pamh, message: str):
    """
    Display an informational message to the user via PAM.

    This is for one-way communication (no response expected).

    Args:
        pamh: PAM handle
        message: The message to display
    """
    try:
        msg = pamh.Message(pamh.PAM_TEXT_INFO, message)
        pamh.conversation(msg)
    except pamh.exception:
        pass  # Best effort


def pam_error(pamh, message: str):
    """
    Display an error message to the user via PAM.

    Args:
        pamh: PAM handle
        message: The error message to display
    """
    try:
        msg = pamh.Message(pamh.PAM_ERROR_MSG, message)
        pamh.conversation(msg)
    except pamh.exception:
        pass  # Best effort


# =============================================================================
# MAIN PAM ENTRY POINTS
# =============================================================================


def pam_sm_authenticate(pamh, flags, argv):
    """
    PAM authentication entry point.

    This function is called by PAM when authentication is requested.
    It orchestrates the entire 2FA flow:

    1. Load configuration
    2. Get user and remote host info
    3. Check bypass conditions
    4. Generate and send OTP code
    5. Prompt user for code
    6. Validate response
    7. Return success or failure

    Args:
        pamh: PAM handle with user info and conversation functions
        flags: PAM flags (not used)
        argv: Module arguments from PAM config (not used)

    Returns:
        PAM_SUCCESS: Authentication successful
        PAM_AUTH_ERR: Authentication failed
        PAM_AUTHINFO_UNAVAIL: Could not send notification
        PAM_IGNORE: 2FA bypassed (user/network in bypass list)
    """
    # -------------------------------------------------------------------------
    # Step 1: Load configuration
    # -------------------------------------------------------------------------
    config = Config(CONFIG_FILE)

    # Initialize logger
    logger = PAMLogger(
        config.get("log_file", "/var/log/pam-ssh-2fa.log"),
        debug=config.get("debug", False),
    )

    logger.debug("PAM authenticate called")

    # -------------------------------------------------------------------------
    # Step 2: Get user and session information
    # -------------------------------------------------------------------------
    try:
        user = pamh.get_user(None)
    except pamh.exception:
        logger.error("Failed to get username from PAM")
        return PAM_USER_UNKNOWN

    if not user:
        logger.error("Empty username")
        return PAM_USER_UNKNOWN

    # Get remote host from PAM environment
    # This is set by sshd when a user connects
    try:
        rhost = pamh.rhost or "unknown"
    except AttributeError:
        rhost = "unknown"

    logger.info("Authentication request", user=user, rhost=rhost)

    # -------------------------------------------------------------------------
    # Step 3: Check bypass conditions
    # -------------------------------------------------------------------------
    bypass_checker = BypassChecker(
        bypass_users=config.get_list("bypass_users"),
        bypass_networks=config.get_list("bypass_networks"),
        logger=logger,
    )

    if bypass_checker.should_bypass(user, rhost):
        return PAM_IGNORE  # Skip 2FA, continue to next PAM module

    # -------------------------------------------------------------------------
    # Step 3.5: Load user-specific configuration
    # -------------------------------------------------------------------------
    # Check if this user has a personal notification config
    # This allows different users to receive codes via different services
    user_config = config.get_user_config(user)
    has_personal_config = config.has_user_config(user)

    if has_personal_config:
        logger.debug("Using per-user config", user=user)

    # -------------------------------------------------------------------------
    # Step 3.6: Check if user has notification URLs configured
    # -------------------------------------------------------------------------
    # User needs either a per-user config with URLs or global URLs must be set
    user_notification_urls = user_config.get_list("apprise_urls")

    if not user_notification_urls:
        # No notification URLs for this user - check allow_unconfigured_users
        allow_unconfigured = config.get("allow_unconfigured_users", False)

        if allow_unconfigured:
            # Allow unconfigured users to bypass 2FA
            logger.info(
                "User has no notification config - bypassing 2FA (allow_unconfigured_users=true)",
                user=user,
                rhost=rhost,
            )
            return PAM_SUCCESS
        else:
            # Deny unconfigured users
            logger.warning(
                "User denied - no notification URLs configured (allow_unconfigured_users=false)",
                user=user,
                rhost=rhost,
            )
            pam_error(pamh, "2FA not configured for this user. Contact administrator.")
            return PAM_AUTH_ERR

    # -------------------------------------------------------------------------
    # Step 4: Get auth method for this user
    # -------------------------------------------------------------------------
    auth_method = user_config.get("auth_method", "code").lower()

    # Validate auth_method
    if auth_method not in ("code", "link", "both", "none"):
        logger.warning(f"Invalid auth_method '{auth_method}', defaulting to 'code'")
        auth_method = "code"

    logger.debug("Auth method selected", user=user, auth_method=auth_method)

    # Handle "none" - skip 2FA for this user
    if auth_method == "none":
        logger.info("2FA skipped - auth_method=none", user=user, rhost=rhost)
        return PAM_SUCCESS

    # -------------------------------------------------------------------------
    # Step 5: Setup managers and generate credentials
    # -------------------------------------------------------------------------
    code_timeout = config.get("code_timeout", 300)
    code = None
    approval_token = None
    approval_link = None

    # Code manager (for "code" and "both" methods)
    code_manager = None
    if auth_method in ("code", "both"):
        code_manager = CodeManager(
            storage_dir=config.get("code_storage_dir", "/var/run/pam-ssh-2fa"),
            code_length=config.get("code_length", 4),
            timeout=code_timeout,
            max_attempts=config.get("max_attempts", 3),
            logger=logger,
        )

        try:
            code = code_manager.generate(user, rhost)
        except OSError as e:
            logger.error(f"Failed to generate code: {e}")
            pam_error(pamh, "Internal error. Please contact administrator.")
            return PAM_AUTHINFO_UNAVAIL

    # Approval manager (for "link" and "both" methods)
    approval_manager = None
    if auth_method in ("link", "both"):
        server_url = config.get("server_url", "")

        if not server_url:
            logger.error("server_url not configured but auth_method requires it")
            pam_error(
                pamh, "2FA link authentication not configured. Contact administrator."
            )
            return PAM_AUTHINFO_UNAVAIL

        approvals_dir = os.path.join(
            config.get("code_storage_dir", "/var/run/pam-ssh-2fa"), "approvals"
        )

        approval_manager = ApprovalManager(
            approvals_dir=approvals_dir,
            timeout=code_timeout,
            server_url=server_url,
            logger=logger,
        )

        try:
            approval_token, approval_link = approval_manager.create_approval(
                user, rhost
            )
        except (OSError, ValueError) as e:
            logger.error(f"Failed to create approval: {e}")
            pam_error(pamh, "Internal error. Please contact administrator.")
            return PAM_AUTHINFO_UNAVAIL

    # -------------------------------------------------------------------------
    # Step 6: Send notification
    # -------------------------------------------------------------------------
    sender = NotificationSender(
        apprise_urls=user_config.get_list("apprise_urls"),
        title_template=user_config.get("notification_title", "SSH Login"),
        body_template=user_config.get("notification_body", "Your code: {code}"),
        body_template_link=user_config.get(
            "notification_body_link", "Click to approve: {link}"
        ),
        body_template_both=user_config.get(
            "notification_body_both", "Click: {link}\n\nOr code: {code}"
        ),
        logger=logger,
    )

    if not sender.send(
        code=code or "",
        user=user,
        rhost=rhost,
        timeout=code_timeout,
        link=approval_link,
        auth_method=auth_method,
    ):
        logger.error("Failed to send notification", user=user, rhost=rhost)
        pam_error(pamh, "Could not send verification. Please try again.")
        # Cleanup approval if created
        if approval_manager and approval_token:
            approval_manager.cleanup(approval_token)
        return PAM_AUTHINFO_UNAVAIL

    # -------------------------------------------------------------------------
    # Step 7: Authenticate based on method
    # -------------------------------------------------------------------------

    if auth_method == "link":
        # -----------------------------------------------------------------
        # Link-only authentication: poll for approval
        # -----------------------------------------------------------------
        pam_info(pamh, "Approval link sent to your device. Waiting for approval...")

        poll_interval = 1.0  # Check every second
        max_wait = code_timeout
        start_time = time.time()

        while True:
            # Check if approved
            if approval_manager.is_approved(approval_token):
                pam_info(
                    pamh, config.get("success_message", "Verification successful.")
                )
                logger.info("Authentication successful (link)", user=user, rhost=rhost)
                approval_manager.cleanup(approval_token)
                return PAM_SUCCESS

            # Check if expired
            if approval_manager.is_expired(approval_token):
                logger.warning("Approval expired", user=user, rhost=rhost)
                pam_error(pamh, config.get("expired_message", "Approval link expired."))
                return PAM_AUTH_ERR

            # Check overall timeout
            if time.time() - start_time > max_wait:
                logger.warning("Approval timeout", user=user, rhost=rhost)
                pam_error(pamh, "Approval timeout. Please try again.")
                approval_manager.cleanup(approval_token)
                return PAM_AUTH_ERR

            time.sleep(poll_interval)

    elif auth_method == "both":
        # -----------------------------------------------------------------
        # Combined authentication: code OR link
        # -----------------------------------------------------------------
        prompt_msg = config.get(
            "prompt_message_both", "Enter code OR press Enter after clicking link: "
        )
        max_attempts = config.get("max_attempts", 3)

        for attempt in range(max_attempts):
            # Check if already approved via link before prompting
            if approval_manager.is_approved(approval_token):
                pam_info(
                    pamh, config.get("success_message", "Verification successful.")
                )
                logger.info("Authentication successful (link)", user=user, rhost=rhost)
                approval_manager.cleanup(approval_token)
                return PAM_SUCCESS

            entered_code = pam_prompt(pamh, prompt_msg, echo=True)

            if entered_code is None:
                # User cancelled or connection issue
                logger.warning("User cancelled authentication", user=user, rhost=rhost)
                approval_manager.cleanup(approval_token)
                return PAM_AUTH_ERR

            # Empty input - check if link was clicked
            if entered_code.strip() == "":
                if approval_manager.is_approved(approval_token):
                    pam_info(
                        pamh, config.get("success_message", "Verification successful.")
                    )
                    logger.info(
                        "Authentication successful (link)", user=user, rhost=rhost
                    )
                    approval_manager.cleanup(approval_token)
                    return PAM_SUCCESS
                else:
                    if attempt < max_attempts - 1:
                        pam_error(
                            pamh, "Link not yet clicked. Enter code or try again."
                        )
                    continue

            # Non-empty input - validate as code
            valid, message = code_manager.validate(user, rhost, entered_code)

            if valid:
                pam_info(
                    pamh, config.get("success_message", "Verification successful.")
                )
                logger.info("Authentication successful (code)", user=user, rhost=rhost)
                approval_manager.cleanup(approval_token)
                return PAM_SUCCESS
            else:
                # Check if link was approved while they were typing
                if approval_manager.is_approved(approval_token):
                    pam_info(
                        pamh, config.get("success_message", "Verification successful.")
                    )
                    logger.info(
                        "Authentication successful (link)", user=user, rhost=rhost
                    )
                    approval_manager.cleanup(approval_token)
                    return PAM_SUCCESS

                if attempt < max_attempts - 1:
                    pam_error(pamh, message)
                else:
                    pam_error(
                        pamh, config.get("failure_message", "Verification failed.")
                    )

        # All attempts exhausted
        logger.warning("Authentication failed - max attempts", user=user, rhost=rhost)
        approval_manager.cleanup(approval_token)
        return PAM_AUTH_ERR

    else:
        # -----------------------------------------------------------------
        # Code-only authentication (default)
        # -----------------------------------------------------------------
        prompt_msg = config.get("prompt_message", "Enter verification code: ")
        max_attempts = config.get("max_attempts", 3)

        for attempt in range(max_attempts):
            entered_code = pam_prompt(pamh, prompt_msg, echo=True)

            if entered_code is None:
                # User cancelled or connection issue
                logger.warning("User cancelled authentication", user=user, rhost=rhost)
                return PAM_AUTH_ERR

            # Validate the code
            valid, message = code_manager.validate(user, rhost, entered_code)

            if valid:
                pam_info(
                    pamh, config.get("success_message", "Verification successful.")
                )
                logger.info("Authentication successful", user=user, rhost=rhost)
                return PAM_SUCCESS
            else:
                # Show error but continue if attempts remain
                if attempt < max_attempts - 1:
                    pam_error(pamh, message)
                else:
                    pam_error(
                        pamh, config.get("failure_message", "Verification failed.")
                    )

        # All attempts exhausted
        logger.warning("Authentication failed - max attempts", user=user, rhost=rhost)
        return PAM_AUTH_ERR


def pam_sm_setcred(pamh, flags, argv):
    """
    PAM set credentials entry point.

    Called by PAM to set user credentials. This module doesn't manage
    credentials, so we just return success.

    Args:
        pamh: PAM handle
        flags: PAM flags
        argv: Module arguments

    Returns:
        PAM_SUCCESS always
    """
    return PAM_SUCCESS


def pam_sm_acct_mgmt(pamh, flags, argv):
    """
    PAM account management entry point.

    Called by PAM to check if account is valid. This module doesn't
    manage accounts, so we return success.

    Args:
        pamh: PAM handle
        flags: PAM flags
        argv: Module arguments

    Returns:
        PAM_SUCCESS always
    """
    return PAM_SUCCESS


def pam_sm_open_session(pamh, flags, argv):
    """
    PAM open session entry point.

    Called when a session is opened. Could be used for logging.

    Args:
        pamh: PAM handle
        flags: PAM flags
        argv: Module arguments

    Returns:
        PAM_SUCCESS always
    """
    return PAM_SUCCESS


def pam_sm_close_session(pamh, flags, argv):
    """
    PAM close session entry point.

    Called when a session is closed. Could be used for cleanup.

    Args:
        pamh: PAM handle
        flags: PAM flags
        argv: Module arguments

    Returns:
        PAM_SUCCESS always
    """
    return PAM_SUCCESS


def pam_sm_chauthtok(pamh, flags, argv):
    """
    PAM change authentication token entry point.

    Called when changing passwords. This module doesn't handle
    password changes.

    Args:
        pamh: PAM handle
        flags: PAM flags
        argv: Module arguments

    Returns:
        PAM_SUCCESS always
    """
    return PAM_SUCCESS


# =============================================================================
# STANDALONE TESTING
# =============================================================================

if __name__ == "__main__":
    """
    Standalone testing mode.

    When run directly (not via PAM), this performs basic tests:
    - Configuration loading
    - Code generation and validation
    - Notification sending (if configured)

    Usage:
        python3 pam_ssh_2fa.py [--test-notify]

        --test-notify: Actually send a test notification
    """
    import argparse

    parser = argparse.ArgumentParser(
        description="PAM SSH 2FA Module - Standalone Testing"
    )
    parser.add_argument(
        "--test-notify", action="store_true", help="Send a test notification"
    )
    parser.add_argument("--config", default=CONFIG_FILE, help="Path to config file")
    args = parser.parse_args()

    print("PAM SSH 2FA Module - Standalone Test")
    print("=" * 40)

    # Test configuration loading
    print("\n[1] Testing configuration loading...")
    try:
        config = Config(args.config)
        print(f"    Config file: {args.config}")
        print(f"    Debug mode: {config.get('debug')}")
        print(f"    Code length: {config.get('code_length')}")
        print(f"    Code timeout: {config.get('code_timeout')}s")
        print(f"    Apprise URLs: {len(config.get_list('apprise_urls'))} configured")
        print("    [OK] Configuration OK")
    except Exception as e:
        print(f"    [FAIL] Configuration error: {e}")
        sys.exit(1)

    # Test logger
    print("\n[2] Testing logger...")
    try:
        logger = PAMLogger("/tmp/pam-ssh-2fa-test.log", debug=True)
        logger.info("Test log entry", test="value")
        print("    [OK] Logger OK")
    except Exception as e:
        print(f"    [FAIL] Logger error: {e}")

    # Test code generation
    print("\n[3] Testing code generation...")
    try:
        code_manager = CodeManager(
            storage_dir="/tmp/pam-ssh-2fa-test",
            code_length=4,
            timeout=60,
            max_attempts=3,
            logger=logger,
        )
        test_code = code_manager.generate("testuser", "127.0.0.1")
        print(f"    Generated code: {test_code}")

        # Test validation with correct code
        valid, msg = code_manager.validate("testuser", "127.0.0.1", test_code)
        if valid:
            print("    [OK] Code validation OK")
        else:
            print(f"    [FAIL] Code validation failed: {msg}")

        # Test validation with incorrect code
        test_code_2 = code_manager.generate("testuser", "127.0.0.1")
        valid, msg = code_manager.validate("testuser", "127.0.0.1", "000000")
        if not valid:
            print("    [OK] Invalid code rejection OK")
        else:
            print("    [FAIL] Invalid code was accepted!")

    except Exception as e:
        print(f"    [FAIL] Code generation error: {e}")

    # Test notification (optional)
    if args.test_notify:
        print("\n[4] Testing notification...")
        urls = config.get_list("apprise_urls")
        if not urls:
            print("    [FAIL] No notification URLs configured")
        else:
            sender = NotificationSender(
                apprise_urls=urls,
                title_template=config.get("notification_title"),
                body_template=config.get("notification_body"),
                logger=logger,
            )
            if sender.send(
                code="1234", user="testuser", rhost="127.0.0.1", timeout=300
            ):
                print("    [OK] Notification sent")
            else:
                print("    [FAIL] Notification failed")

    # Cleanup
    print("\n[5] Cleaning up test files...")
    import shutil

    try:
        shutil.rmtree("/tmp/pam-ssh-2fa-test", ignore_errors=True)
        os.unlink("/tmp/pam-ssh-2fa-test.log")
    except OSError:
        pass
    print("    [OK] Cleanup done")

    print("\n" + "=" * 40)
    print("Testing complete!")