diff --git a/README.md b/README.md index 112a146..cc2980c 100644 --- a/README.md +++ b/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 diff --git a/docs/application_development/configuration.md b/docs/application_development/configuration.md index a956594..770b38e 100644 --- a/docs/application_development/configuration.md +++ b/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 | diff --git a/docs/application_development/security.md b/docs/application_development/security.md index be512c9..852fcad 100644 --- a/docs/application_development/security.md +++ b/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: + +``` +. +``` + +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 +
WWW-Authenticate] H -->|Yes| J[Authenticate] J --> K{Valid credentials?} - K -->|No| L[401 +
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 diff --git a/src/pyrobusta/protocol/http.py b/src/pyrobusta/protocol/http.py index 225550f..50bc356 100644 --- a/src/pyrobusta/protocol/http.py +++ b/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: """ diff --git a/src/pyrobusta/protocol/http_basic_auth.py b/src/pyrobusta/protocol/http_basic_auth.py index e80c482..587bd8d 100644 --- a/src/pyrobusta/protocol/http_basic_auth.py +++ b/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"", b""]} +_USERS = {} # {"user": [b"", b"", b""]} _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) diff --git a/src/pyrobusta/utils/config.py b/src/pyrobusta/utils/config.py index 8ae559a..d11045a 100644 --- a/src/pyrobusta/utils/config.py +++ b/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, diff --git a/src/pyrobusta/utils/crypto.py b/src/pyrobusta/utils/crypto.py new file mode 100644 index 0000000..87490f0 --- /dev/null +++ b/src/pyrobusta/utils/crypto.py @@ -0,0 +1,77 @@ +""" +Utility functions for cryptography operations. +""" + +import binascii +import hashlib +import os + + +class HmacSha256: + # pylint: disable=R0903 + """ + Helper class for SHA-256-based Hashed MAC (HMAC) + computation, based on RFC 2104. + """ + + def __init__(self, key): + if len(key) > 64: + key = hashlib.sha256(key).digest() + if len(key) < 64: + key = key + b"\x00" * (64 - len(key)) + self.ipad = bytes(x ^ 0x36 for x in key) + self.opad = bytes(x ^ 0x5C for x in key) + + def digest(self, msg): + """ + Calculate the HMAC digest of a message. + """ + inner = hashlib.sha256() + inner.update(self.ipad) + inner.update(msg) + outer = hashlib.sha256() + outer.update(self.opad) + outer.update(inner.digest()) + return outer.digest() + + +def constant_time_equal(a, b): + """ + Constant time comparison 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 create_signed_token(secret, nonce_size=16): + """ + Create a signed token containing only a random nonce. + """ + nonce = os.urandom(nonce_size) + data = nonce + hmac = HmacSha256(secret) + signature = binascii.hexlify(hmac.digest(data)) + return binascii.hexlify(data) + b"." + signature + + +def verify_signed_token(secret, token, nonce_size=16): + """ + Verify a signed token containing only a random nonce. + """ + try: + sep = token.find(".") + if sep == -1: + return False + data = binascii.unhexlify(token[:sep]) + if len(data) != nonce_size: + return False + request_signature = binascii.unhexlify(token[sep + 1 :]) + hmac = HmacSha256(secret) + expected_signature = hmac.digest(data) + except (ValueError, binascii.Error): + return False + return constant_time_equal(request_signature, expected_signature) diff --git a/tests/unit/http_base.py b/tests/unit/http_base.py index 7756d50..4cd4645 100644 --- a/tests/unit/http_base.py +++ b/tests/unit/http_base.py @@ -73,6 +73,7 @@ def setUp(self): # ------------------------------------------------ # Load remaining modules, enable optional features # ------------------------------------------------ + self.crypto_module = load_module("pyrobusta/utils/crypto.py") self.http_module = load_module("pyrobusta/protocol/http.py") self.fs_module = load_module("pyrobusta/protocol/http_file_server.py") self.multipart_module = load_module("pyrobusta/protocol/http_multipart.py") diff --git a/tests/unit/test_http_basic_auth.py b/tests/unit/test_http_basic_auth.py index 71ad2be..504f96d 100644 --- a/tests/unit/test_http_basic_auth.py +++ b/tests/unit/test_http_basic_auth.py @@ -73,7 +73,12 @@ def setUpClass(cls): def prepare_auth(self, user, password, auth_header, user_roles, route_roles): if user is not None and password is not None: password_hash = hashlib.sha256(password.encode()).digest() - self.basic_auth_module._USERS[user] = [password_hash, user_roles] + nonce_size = self.basic_auth_module._CSRF_USER_SECRET_SIZE + self.basic_auth_module._USERS[user] = [ + password_hash, + user_roles, + os.urandom(nonce_size), + ] self.basic_auth_module._ATTR_TREE.insert_path( "/app/private", {"GET": route_roles} @@ -306,6 +311,216 @@ def test_basic_auth_invalid_scheme(self): self.assertEqual(self.engine.status_code, 401) +class TestBasicAuthCSRFStateMachine(TestHttpBase): + """ + Tests for HTTP authentication with CSRF protection. + """ + + @classmethod + def setUpClass(cls): + cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + # For disbaling authentication related warning messages + "tls": True, + } + + def prepare_auth(self, user, password, auth_header, user_roles, route_roles): + if user is not None and password is not None: + password_hash = hashlib.sha256(password.encode()).digest() + nonce_size = self.basic_auth_module._CSRF_USER_SECRET_SIZE + self.basic_auth_module._USERS[user] = [ + password_hash, + user_roles, + os.urandom(nonce_size), + ] + + self.basic_auth_module._ATTR_TREE.insert_path( + "/app/private", {"GET": route_roles, "POST": route_roles} + ) + self.engine.state = self.engine._handle_auth_header_st + self.engine.url = b"/app/private" + if auth_header: + self.engine.headers["authorization"] = auth_header + + def test_csrf_generated_token_valid(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + self.engine.method = b"GET" + + self.engine.state(self.rx) + + user_secret = self.basic_auth_module._USERS["dummy-user"][2] + cookie_name, csrf_token = ( + self.engine._lookup(self.engine.resp_headers, b"set-cookie") + .decode("ascii") + .split(";")[0] + .split("=") + ) + is_token_valid = self.crypto_module.verify_signed_token( + user_secret, csrf_token, self.basic_auth_module._CSRF_NONCE_SIZE + ) + + self.assertEqual(cookie_name, "csrf-token") + self.assertTrue(is_token_valid) + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertEqual(self.engine.status_code, None) + + def test_csrf_existing_token_accepted(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + self.engine.method = b"GET" + + user_secret = self.basic_auth_module._USERS["dummy-user"][2] + csrf_token = self.crypto_module.create_signed_token( + user_secret, self.basic_auth_module._CSRF_NONCE_SIZE + ) + self.engine.headers["cookie"] = "csrf-token=" + csrf_token.decode("ascii") + self.engine.headers["x-csrf-token"] = csrf_token.decode("ascii") + + self.engine.state(self.rx) + + with self.assertRaises(ValueError): + self.engine._lookup(self.engine.resp_headers, b"set-cookie") + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertEqual(self.engine.status_code, None) + + def test_csrf_request_token_accepted(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + self.engine.method = b"POST" + + user_secret = self.basic_auth_module._USERS["dummy-user"][2] + csrf_token = self.crypto_module.create_signed_token( + user_secret, self.basic_auth_module._CSRF_NONCE_SIZE + ) + self.engine.headers["cookie"] = "csrf-token=" + csrf_token.decode("ascii") + self.engine.headers["x-csrf-token"] = csrf_token.decode("ascii") + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertEqual(self.engine.status_code, None) + + def test_csrf_request_forged_token_rejected(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + self.engine.method = b"POST" + + forged_user_secret = os.urandom(self.basic_auth_module._CSRF_NONCE_SIZE) + csrf_token = self.crypto_module.create_signed_token( + forged_user_secret, self.basic_auth_module._CSRF_NONCE_SIZE + ) + self.engine.headers["cookie"] = "csrf-token=" + csrf_token.decode("ascii") + self.engine.headers["x-csrf-token"] = csrf_token.decode("ascii") + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 403) + + def test_csrf_request_missing_csrf_rejected(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + self.engine.method = b"POST" + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 403) + + def test_csrf_missing_cookie_rejected(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + self.engine.method = b"POST" + + user_secret = self.basic_auth_module._USERS["dummy-user"][2] + csrf_token = self.crypto_module.create_signed_token( + user_secret, self.basic_auth_module._CSRF_NONCE_SIZE + ) + self.engine.headers["x-csrf-token"] = csrf_token.decode("ascii") + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 403) + + def test_csrf_missing_token_rejected(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + self.engine.method = b"POST" + + user_secret = self.basic_auth_module._USERS["dummy-user"][2] + csrf_token = self.crypto_module.create_signed_token( + user_secret, self.basic_auth_module._CSRF_NONCE_SIZE + ) + self.engine.headers["cookie"] = "csrf-token=" + csrf_token.decode("ascii") + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 403) + + class TestBasicAuthPrefixTree(TestHttpBase): """ Tests for authorization based on prefix tree. @@ -488,18 +703,21 @@ def test_user_reader_valid_config(self): "70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145" ), 0b00, + self.basic_auth_module._USERS["user-1"][2], ], "user-2": [ bytes.fromhex( "718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf" ), 0b01, + self.basic_auth_module._USERS["user-2"][2], ], "user-3": [ bytes.fromhex( "280b73d2ac8375035501e5c06acb4b770e74ec95f479e6e2643227d7f57ce7ad" ), 0b11, + self.basic_auth_module._USERS["user-3"][2], ], }, )