Add CSRF protection for HTTP Basic Authentication

README.md

@@ -11,6 +11,7 @@ PyRobusta is a memory-conscious HTTP/1.1 server library built for embedded devic
 - Persistent connection handling via the `Connection: keep-alive` header
 - HTTP/1.0 and HTTP/1.1 support
 - Basic authentication and RBAC authorization
+- CSRF protection for browser clients
 - TLS support
 
 ## Design Principles

docs/application_development/configuration.md

@@ -52,6 +52,7 @@ $ mpremote a0 cp pyrobusta.env :/pyrobusta.env
 | `http_served_paths` | Space-separated list of filesystem paths that may be served over HTTP. | `/www /lib/pyrobusta` |
 | `http_files_api` | Enables or disables the file management API endpoint (`/files`), allowing upload, download, and listing of files. | False |
 | `http_auth` | Selects the type of authentication method enforced by the server. Currently, basic authentication (`basic`) is supported. | None |
+| `http_auth_mode` | Selects authentication usage model. `browser` enables CSRF protection; `api` assumes credentials are explicitly provided by the client and disables CSRF protection. | `browser` |
 | `http_insecure_auth` | Allows clients to authenticate over unsecured HTTP (without TLS). This may expose credentials or authentication tokens in transit. | False |
 | `socket_max_con` | Maximum number of simultaneous socket connections. | 2 |
 | `tls` | Enables or disables TLS. When enabled, `cert.der` and `key.der` must be installed at the server root. | False |

docs/application_development/security.md

@@ -158,6 +158,37 @@ and is therefore not supported.
 **Note**: The special role `*` grants public access regardless of the authenticated user's roles.
 Rules that specify `*` do not require authentication.
 
+## CSRF Protection
+
+Credentials used with HTTP Basic Authentication may be automatically attached to requests by a browser.
+This makes HTTP Basic Authentication vulnerable to CSRF (Cross-Site Request Forgery) attacks,
+allowing an attacker-controlled website to cause the browser to issue requests to the server on
+behalf of an authenticated user.
+
+To prevent CSRF attacks, **PyRobusta applies the Signed Double-Submit Cookie pattern**. For unsafe HTTP methods
+(currently POST, PUT, and DELETE), the client must return the CSRF token received from the server as
+a cookie. The client must send the same token in the `X-CSRF-Token` request header. The server
+rejects requests where the cookie value and header value do not match.
+
+A malicious website cannot read the CSRF cookie from another origin because of the browser's
+same-origin policy. Therefore, a malicious website cannot provide a valid CSRF token in the request header.
+Cryptographic signing prevents attackers from forging valid CSRF tokens without knowledge of the server-side secret.
+
+PyRobusta generates CSRF tokens using HMAC-SHA256 with per-user secrets generated during server
+initialization. User secrets exist only in memory, so restarting the server invalidates all
+previous CSRF tokens. The server issues a CSRF cookie after successful authentication. Clients
+must include the same token in the `X-CSRF-Token` header for subsequent unsafe requests.
+
+Tokens generated by PyRobusta have the following format:
+
+```
+<hex-encoded-random-nonce>.<hex-encoded-HMAC-SHA256-signature>
+```
+
+During verification, PyRobusta recalculates the HMAC-SHA256 signature over the nonce using the
+user secret and compares it with the signature provided in the token. Requests are rejected with HTTP
+`403 Forbidden` if the token signature is invalid or if the cookie and `X-CSRF-Token` header values do not match.
+
 ## Authentication & Authorization Flow
 
 ```mermaid
@@ -171,25 +202,31 @@ flowchart TD
     D --> F[Ignore auth requirement]
     F --> G[Process request]
 
-    E --> H{Is Authorization header present?}
+    E --> H{Authorization header present?}
 
     H -->|No| I[401 +<br/>WWW-Authenticate]
     H -->|Yes| J[Authenticate]
 
     J --> K{Valid credentials?}
 
-    K -->|No| L[401 +<br/>WWW-Authenticate]
-    K -->|Yes| M{Authorized?}
+    K -->|No| I
+    K -->|Yes| L{Valid CSRF token?}
+
+    L -->|No| M[403 Forbidden]
+    L -->|Yes| N{Authorized?}
 
-    M -->|No| N[403]
-    M -->|Yes| G
+    N -->|No| M
+    N -->|Yes| G
 ```
 
+PyRobusta returns the following HTTP status codes for authentication and authorization failures:
+
 | Condition | Response |
 | --- | --- |
-| Missing credentials | 401 Unauthorized |
-| Invalid credentials | 401 Unauthorized |
-| Authenticated but insufficient permissions | 403 Forbidden |
+| Missing authentication credentials | 401 Unauthorized |
+| Invalid authentication credentials | 401 Unauthorized |
+| Authenticated, but CSRF validation failed | 403 Forbidden |
+| Authenticated, but authorization failed | 403 Forbidden |
 
 ## HTTPS / TLS
 

src/pyrobusta/protocol/http.py

@@ -293,6 +293,21 @@ def get_query_param(self, key: str, default: str = None) -> str:
             return self.query[idx_start + len(key) + 1 : idx_end]
         return self.query[idx_start + len(key) + 1 :]
 
+    def get_cookie(self, name, default=None):
+        """
+        Obtain a named cookie from the request headers.
+        """
+        cookie_header = self.headers.get("cookie", "")
+        name = name.lower()
+        for part in helpers.iterate_segments(cookie_header, ";"):
+            cookie_sep = part.find("=")
+            if cookie_sep == -1:
+                continue
+            key = part[:cookie_sep].strip()
+            if key.lower() == name:
+                return part[cookie_sep + 1 :].strip()
+        return default
+
     @staticmethod
     def _is_matching_url_path(path: bytes, pattern: bytes) -> bool:
         """

src/pyrobusta/protocol/http_basic_auth.py

@@ -10,16 +10,23 @@
 
 import hashlib
 import gc
+import os
 import binascii
 
 from pyrobusta.protocol import http
 from pyrobusta.utils.patch import add_method
 from pyrobusta.utils.helpers import iterate_segments
+from pyrobusta.utils.crypto import (
+    constant_time_equal,
+    create_signed_token,
+    verify_signed_token,
+)
 from pyrobusta.utils.config import (
     get_config,
     CONF_PASSWD_FILE,
     CONF_ROLES_FILE,
     CONF_HTTP_AUTH,
+    CONF_HTTP_AUTH_MODE,
     CONF_HTTP_INSECURE_AUTH,
     CONF_TLS,
 )
@@ -30,9 +37,11 @@
 _ALL_ROLES = 2**_MAX_ROLES - 1
 
 _ROLE_INDEX = {}
-_USERS = {}  # {"user": [b"<password-hash>", b"<role-mask>"]}
+_USERS = {}  # {"user": [b"<password-hash>", b"<role-mask>", b"<csrf_signing_key>"]}
 _ATTR_TREE = None  # root node: /
 
+_CSRF_NONCE_SIZE = 16
+_CSRF_USER_SECRET_SIZE = 32
 _DUMMY_HASH = hashlib.sha256(b"invalid-user").digest()
 
 
@@ -192,7 +201,11 @@ def _load_users():
                     raise ValueError()
                 if user in _USERS:
                     raise ValueError()
-                _USERS[user] = [bytes.fromhex(password_hash), role_mask]
+                _USERS[user] = [
+                    bytes.fromhex(password_hash),
+                    role_mask,
+                    os.urandom(_CSRF_USER_SECRET_SIZE),
+                ]
     except OSError:
         warning("Unable to open: " + passwd_file)
 
@@ -260,46 +273,22 @@ def _handle_auth_st(self, _):
         self.state = self._handle_auth_header_st
 
 
-def compare_digest(a, b):
-    """
-    Constant time comparison of hash digests to prevent timing attacks.
-    """
-    if len(a) != len(b):
-        return False
-    diff = 0
-    for x, y in zip(a, b):
-        diff |= x ^ y
-    return diff == 0
-
-
-def _handle_auth_header_st(self, _):
-    method = self.method.decode("ascii")
-    url = self.url.decode("ascii")
-    auth_header = self.headers.get("authorization", "").strip()
-    www_auth_header = b"WWW-Authenticate"
-    www_auth_method = b'Basic realm="Device"'
-
+def _authenticate(auth_header):
     # Protocol validation
-    if not auth_header or auth_header.strip()[:5].lower() != "basic":
-        self.set_response_header(www_auth_header, www_auth_method)
-        self.terminate(401)
-        return
+    if not auth_header or auth_header[:6].lower() != "basic ":
+        return None
 
     # Decoding
-    user_data = auth_header[5:].strip()
+    user_data = auth_header[6:].strip()
     try:
         user_data = binascii.a2b_base64(user_data).decode()
     except binascii.Error:
-        self.set_response_header(www_auth_header, www_auth_method)
-        self.terminate(401)
-        return
+        return None
 
     # Authentication
     user_sep = user_data.find(":")
     if user_sep < 0:
-        self.set_response_header(www_auth_header, www_auth_method)
-        self.terminate(401)
-        return
+        return None
 
     username = user_data[:user_sep].lower()
     user_info = _USERS.get(username)
@@ -308,14 +297,61 @@ def _handle_auth_header_st(self, _):
         user_data[user_sep + 1 :].strip().encode("ascii")
     ).digest()
 
-    hash_ok = compare_digest(password_hash, stored_hash)
+    hash_ok = constant_time_equal(password_hash, stored_hash)
     user_ok = user_info is not None
 
     if not (user_ok and hash_ok):
-        self.set_response_header(www_auth_header, www_auth_method)
+        return None
+
+    return user_info
+
+
+def _is_valid_csrf_token(cookie_token, header_token, user_secret):
+    if cookie_token is None or header_token is None:
+        return False
+    csrf_sep = header_token.find(".")
+    if csrf_sep == -1:
+        return False
+    if cookie_token != header_token:
+        return False
+    if not verify_signed_token(user_secret, header_token, _CSRF_NONCE_SIZE):
+        return False
+    return True
+
+
+def _handle_auth_header_st(self, _):
+    method = self.method.decode("ascii")
+    url = self.url.decode("ascii")
+    auth_header = self.headers.get("authorization", "").strip()
+
+    # Authentication
+    if not (user_info := _authenticate(auth_header)):
+        self.set_response_header(b"WWW-Authenticate", b'Basic realm="Device"')
         self.terminate(401)
         return
 
+    # CSRF validation, cookie setting
+    if get_config(CONF_HTTP_AUTH_MODE) == "browser":
+        if self.method not in (
+            self.GET,
+            self.HEAD,
+            self.OPTIONS,
+        ):
+            if not _is_valid_csrf_token(
+                self.get_cookie("csrf-token"),
+                self.headers.get("x-csrf-token"),
+                user_info[2],
+            ):
+                self.terminate(403)
+                return
+        elif self.method in (self.GET, self.HEAD):
+            if self.get_cookie("csrf-token") is None:
+                csrf_token = create_signed_token(user_info[2], _CSRF_NONCE_SIZE)
+                cookie = b"csrf-token=" + csrf_token + b"; path=/; samesite=strict"
+                if get_config(CONF_TLS):
+                    cookie += b"; secure"
+                self.set_response_header(b"set-cookie", cookie)
+
     # Authorization
     policy = _ATTR_TREE.get_attributes(url)
 
@@ -346,6 +382,12 @@ def apply_patches():
             else:
                 raise ValueError(insecure_auth_msg)
 
+        if get_config(CONF_HTTP_AUTH_MODE) != "browser":
+            warning(
+                "CSRF protection is disabled; authenticated browser "
+                "requests may be vulnerable to cross-site request forgery"
+            )
+
         add_method(http.HttpEngine, _handle_auth_st)
         add_method(http.HttpEngine, _handle_auth_header_st)
 

src/pyrobusta/utils/config.py

@@ -30,12 +30,13 @@ def const(n):  # pylint: disable=C0116
 CONF_HTTP_SERVED_PATHS = const(6)
 CONF_HTTP_FILES_API = const(7)
 CONF_HTTP_AUTH = const(8)
-CONF_HTTP_INSECURE_AUTH = const(9)
-CONF_SOCKET_MAX_CON = const(10)
-CONF_TLS = const(11)
-CONF_LOG_LEVEL = const(12)
-CONF_PASSWD_FILE = const(13)
-CONF_ROLES_FILE = const(14)
+CONF_HTTP_AUTH_MODE = const(9)
+CONF_HTTP_INSECURE_AUTH = const(10)
+CONF_SOCKET_MAX_CON = const(11)
+CONF_TLS = const(12)
+CONF_LOG_LEVEL = const(13)
+CONF_PASSWD_FILE = const(14)
+CONF_ROLES_FILE = const(15)
 
 # -------------------
 # Configuration state
@@ -60,6 +61,8 @@ def const(n):  # pylint: disable=C0116
     False,
     CONF_HTTP_AUTH,
     None,
+    CONF_HTTP_AUTH_MODE,
+    "browser",
     CONF_HTTP_INSECURE_AUTH,
     False,
     CONF_SOCKET_MAX_CON,