Add PATCH support and improve configuration, typing, and path handling

Makefile

@@ -281,6 +281,7 @@ stage-test:
 
 	@cp -r build/pyrobusta $(TEST_RUNTIME)/lib
 	@cp tests/functional/*.py $(TEST_RUNTIME)/
+	@cp -r dist/pyrobusta/assets $(TEST_RUNTIME)/lib/pyrobusta
 
 # -----------------------------
 # Run functional tests on UNIX port

docs/application_development/configuration.md

@@ -56,6 +56,8 @@ $ mpremote a0 cp pyrobusta.env :/pyrobusta.env
 | `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 |
+| `tls_cert_file` | Path to the TLS certificate. | `/cert.der` |
+| `tls_key_file` | Path to the TLS private key. | `/key.der` |
 | `passwd_file` | Path to the file containing user credentials used for authentication. | `/pyrobusta.passwd` |
 | `roles_file` | Path to the file containing RBAC role definitions used for authorization. | `/pyrobusta.roles` |
 | `log_level` | Logging level. Can be one of: `warning`, `info`, `debug`. | `info` |

docs/application_development/security.md

@@ -166,7 +166,7 @@ allowing an attacker-controlled website to cause the browser to issue requests t
 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
+(POST, PUT, PATCH, 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.
 

src/pyrobusta/protocol/http.py

@@ -133,9 +133,10 @@ class HttpEngine:
     GET = b"GET"
     HEAD = b"HEAD"
     OPTIONS = b"OPTIONS"
+    PATCH = b"PATCH"
     POST = b"POST"
     PUT = b"PUT"
-    METHODS = (DELETE, GET, HEAD, OPTIONS, POST, PUT)
+    METHODS = (DELETE, GET, HEAD, OPTIONS, PATCH, POST, PUT)
     SUPPORTED_VERSIONS = (b"HTTP/1.1", b"HTTP/1.0")
     SESSION_COUNTER = 0
 

src/pyrobusta/protocol/http_basic_auth.py

@@ -273,7 +273,7 @@ def _handle_auth_st(self, _):
         self.state = self._handle_auth_header_st
 
 
-def _authenticate(auth_header):
+def _authenticate(auth_header: str):
     # Protocol validation
     if not auth_header or auth_header[:6].lower() != "basic ":
         return None
@@ -306,10 +306,10 @@ def _authenticate(auth_header):
     return user_info
 
 
-def _is_valid_csrf_token(cookie_token, header_token, user_secret):
-    if cookie_token is None or header_token is None:
+def _is_valid_csrf_token(cookie_token: bytes, header_token: bytes, user_secret: bytes):
+    if not cookie_token or not header_token:
         return False
-    csrf_sep = header_token.find(".")
+    csrf_sep = header_token.find(b".")
     if csrf_sep == -1:
         return False
     if cookie_token != header_token:
@@ -338,8 +338,8 @@ def _handle_auth_header_st(self, _):
             self.OPTIONS,
         ):
             if not _is_valid_csrf_token(
-                self.get_cookie("csrf-token"),
-                self.headers.get("x-csrf-token"),
+                self.get_cookie("csrf-token", "").encode("ascii"),
+                self.headers.get("x-csrf-token", "").encode("ascii"),
                 user_info[2],
             ):
                 self.terminate(403)

src/pyrobusta/server/http_server.py

@@ -15,9 +15,10 @@
     CONF_HTTPS_PORT,
     CONF_HTTP_MEM_CAP,
     CONF_TLS,
+    CONF_TLS_CERT_FILE,
+    CONF_TLS_KEY_FILE,
     CONF_SOCKET_MAX_CON,
 )
-from ..utils.lexpath import normalize_path
 from ..utils import logging
 
 
@@ -40,8 +41,6 @@ class HttpServer:
     )
     LISTEN_PORT_HTTP = get_config(CONF_HTTP_PORT)
     LISTEN_PORT_HTTPS = get_config(CONF_HTTPS_PORT)
-    TLS_CERT_PATH = "/cert.der"
-    TLS_KEY_PATH = "/key.der"
     CON_TIMEOUT_S = 30
 
     # -----------------------------------------
@@ -185,8 +184,8 @@ async def start_socket_server(self):
 
                 ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
                 ssl_ctx.load_cert_chain(
-                    normalize_path(self.TLS_CERT_PATH),
-                    normalize_path(self.TLS_KEY_PATH),
+                    get_config(CONF_TLS_CERT_FILE),
+                    get_config(CONF_TLS_KEY_FILE),
                 )
 
             self._server = await start_server(

src/pyrobusta/utils/assets.py

@@ -66,7 +66,7 @@ def install_www():
     """
     source_dir = normalize_path("/lib/pyrobusta/assets/www")
     target_dir = normalize_path("/www")
-    if "www" not in listdir():
+    if "www" not in listdir(normalize_path("/")):
         mkdir(target_dir)
 
     for asset_dir in iterate_fs(source_dir, FS_ITER_DIR, FS_ITER_ABS):

src/pyrobusta/utils/config.py

@@ -34,47 +34,38 @@ def const(n):  # pylint: disable=C0116
 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)
+CONF_TLS_CERT_FILE = const(13)
+CONF_TLS_KEY_FILE = const(14)
+CONF_PASSWD_FILE = const(15)
+CONF_ROLES_FILE = const(16)
+CONF_LOG_LEVEL = const(17)
 
 # -------------------
 # Configuration state
 # -------------------
 _CONFIG_LOADED = False
 _CONFIG_CACHE = [
-    CONF_WIFI_SSID,
-    None,
-    CONF_WIFI_PASSWORD,
-    None,
-    CONF_HTTP_PORT,
-    80,
-    CONF_HTTPS_PORT,
-    443,
-    CONF_HTTP_MULTIPART,
-    False,
-    CONF_HTTP_MEM_CAP,
-    0.1,
-    CONF_HTTP_SERVED_PATHS,
-    [normalize_path("/www"), normalize_path("/lib/pyrobusta")],
-    CONF_HTTP_FILES_API,
-    False,
-    CONF_HTTP_AUTH,
-    None,
-    CONF_HTTP_AUTH_MODE,
-    "browser",
-    CONF_HTTP_INSECURE_AUTH,
-    False,
-    CONF_SOCKET_MAX_CON,
-    2,
-    CONF_TLS,
-    False,
-    CONF_LOG_LEVEL,
-    "info",
-    CONF_PASSWD_FILE,
-    normalize_path("/pyrobusta.passwd"),
-    CONF_ROLES_FILE,
-    normalize_path("/pyrobusta.roles"),
+    None,  # CONF_WIFI_SSID
+    None,  # CONF_WIFI_PASSWORD
+    80,  # CONF_HTTP_PORT
+    443,  # CONF_HTTPS_PORT
+    False,  # CONF_HTTP_MULTIPART
+    0.1,  # CONF_HTTP_MEM_CAP
+    [
+        normalize_path("/www"),
+        normalize_path("/lib/pyrobusta"),
+    ],  # CONF_HTTP_SERVED_PATHS
+    False,  # CONF_HTTP_FILES_API
+    None,  # CONF_HTTP_AUTH
+    "browser",  # CONF_HTTP_AUTH
+    False,  # CONF_HTTP_INSECURE_AUTH
+    2,  # CONF_SOCKET_MAX_CON
+    False,  # CONF_TLS
+    normalize_path("/cert.der"),  # CONF_TLS_CERT_FILE
+    normalize_path("/key.der"),  # CONF_TLS_KEY_FILE
+    normalize_path("/pyrobusta.passwd"),  # CONF_PASSWD_FILE
+    normalize_path("/pyrobusta.roles"),  # CONF_ROLES_FILE
+    "info",  # CONF_LOG_LEVEL
 ]
 
 
@@ -99,7 +90,12 @@ def parse_config(key, value):
         return float(value)
     if key == CONF_HTTP_SERVED_PATHS:
         return [normalize_path(p) for p in value.split()]
-    if key in (CONF_PASSWD_FILE, CONF_ROLES_FILE):
+    if key in (
+        CONF_PASSWD_FILE,
+        CONF_ROLES_FILE,
+        CONF_TLS_CERT_FILE,
+        CONF_TLS_KEY_FILE,
+    ):
         return normalize_path(value)
     if key in (CONF_WIFI_SSID, CONF_WIFI_PASSWORD):
         return value
@@ -108,32 +104,24 @@ def parse_config(key, value):
 
 def read_config(config=CONFIG_LOCATION):
     """
-    Read configuration from a file and update CONFIG_CACHE.
-    :param config: path to configuration
+    Read configuration from a file and update _CONFIG_CACHE.
+    :param config: path to configuration.
     """
     try:
         with open(config, encoding="utf-8") as conf:
             for line in conf:
                 line = line.rstrip("\r\n").split("#")[0]
                 if not line.strip():
                     continue
-                parts = line.split("=")
-                key_name = "CONF_" + parts[0].strip().upper()
-                if key_name in globals():
-                    key = globals()[key_name]
-                else:
-                    key = len(_CONFIG_CACHE) // 2 + 1
-                    globals()[key_name] = key
-                value = parts[1].strip().strip("'").strip('"')
-                value = parse_config(key, value)
-                if (
-                    key in _CONFIG_CACHE
-                    and (conf_idx := _CONFIG_CACHE.index(key)) % 2 == 0
-                ):
-                    _CONFIG_CACHE[conf_idx + 1] = value
-                else:
-                    _CONFIG_CACHE.append(key)
-                    _CONFIG_CACHE.append(value)
+
+                key_name, value = line.split("=", 1)
+                key = globals().get("CONF_" + key_name.strip().upper())
+                if key is None:
+                    continue
+
+                value = value.strip().strip("'").strip('"')
+                _CONFIG_CACHE[key] = parse_config(key, value)
+
     except OSError:
         pass
 
@@ -147,6 +135,8 @@ def is_protected_file(norm_path: str):
         CONFIG_LOCATION,
         get_config(CONF_PASSWD_FILE),
         get_config(CONF_ROLES_FILE),
+        get_config(CONF_TLS_CERT_FILE),
+        get_config(CONF_TLS_KEY_FILE),
     )
 
 
@@ -159,4 +149,4 @@ def get_config(key):
     if not _CONFIG_LOADED:
         read_config()
         _CONFIG_LOADED = True
-    return _CONFIG_CACHE[2 * key + 1]
+    return _CONFIG_CACHE[key]

src/pyrobusta/utils/crypto.py

@@ -14,15 +14,15 @@ class HmacSha256:
     computation, based on RFC 2104.
     """
 
-    def __init__(self, key):
+    def __init__(self, key: bytes):
         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):
+    def digest(self, msg: bytes):
         """
         Calculate the HMAC digest of a message.
         """
@@ -35,7 +35,7 @@ def digest(self, msg):
         return outer.digest()
 
 
-def constant_time_equal(a, b):
+def constant_time_equal(a: bytes, b: bytes):
     """
     Constant time comparison to prevent timing attacks.
     """
@@ -47,7 +47,7 @@ def constant_time_equal(a, b):
     return diff == 0
 
 
-def create_signed_token(secret, nonce_size=16):
+def create_signed_token(secret: bytes, nonce_size: int = 16):
     """
     Create a signed token containing only a random nonce.
     """
@@ -58,12 +58,12 @@ def create_signed_token(secret, nonce_size=16):
     return binascii.hexlify(data) + b"." + signature
 
 
-def verify_signed_token(secret, token, nonce_size=16):
+def verify_signed_token(secret: bytes, token: bytes, nonce_size: int = 16):
     """
     Verify a signed token containing only a random nonce.
     """
     try:
-        sep = token.find(".")
+        sep = token.find(b".")
         if sep == -1:
             return False
         data = binascii.unhexlify(token[:sep])

tests/functional/env_utils.py

@@ -111,14 +111,14 @@ def setup_config(
     http_server.HttpServer.LISTEN_PORT_HTTP = 8080
     http_server.HttpServer.LISTEN_PORT_HTTPS = 4443
 
-    _CONFIG_CACHE[2 * CONF_LOG_LEVEL + 1] = "warning"
-    _CONFIG_CACHE[2 * CONF_TLS + 1] = tls_enabled
-    _CONFIG_CACHE[2 * CONF_HTTP_SERVED_PATHS + 1] = parse_config(
+    _CONFIG_CACHE[CONF_LOG_LEVEL] = "warning"
+    _CONFIG_CACHE[CONF_TLS] = tls_enabled
+    _CONFIG_CACHE[CONF_HTTP_SERVED_PATHS] = parse_config(
         CONF_HTTP_SERVED_PATHS, served_paths
     )
-    _CONFIG_CACHE[2 * CONF_HTTP_MULTIPART + 1] = http_multipart_enabled
-    _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = files_api_enabled
-    _CONFIG_CACHE[2 * CONF_HTTP_AUTH + 1] = http_auth
-    _CONFIG_CACHE[2 * CONF_HTTP_INSECURE_AUTH + 1] = http_insecure_auth
+    _CONFIG_CACHE[CONF_HTTP_MULTIPART] = http_multipart_enabled
+    _CONFIG_CACHE[CONF_HTTP_FILES_API] = files_api_enabled
+    _CONFIG_CACHE[CONF_HTTP_AUTH] = http_auth
+    _CONFIG_CACHE[CONF_HTTP_INSECURE_AUTH] = http_insecure_auth
 
     enable_optional_features()

tests/functional/test_http.py

@@ -11,9 +11,8 @@
     delete_path,
 )
 
-from pyrobusta.server import http_server
 from pyrobusta.protocol.http import HttpEngine
-
+from pyrobusta.utils.assets import install_www, iterate_fs
 from pyrobusta.utils.lexpath import normalize_path
 
 
@@ -187,7 +186,7 @@ async def test_fs_access_control():
             True,
         )
     finally:
-        delete_path(test_root)
+        delete_path(www_root)
         await server.terminate()
 
 
@@ -275,6 +274,23 @@ async def test_keepalive():
         await server.terminate()
 
 
+def test_www_install():
+    www_root = normalize_path("/www")
+
+    try:
+        install_www()
+        origin_files = set(
+            file.split("/")[-1]
+            for file in iterate_fs(normalize_path("/lib/pyrobusta/assets/www"))
+        )
+        target_files = set(
+            file.split("/")[-1] for file in iterate_fs(normalize_path("/www"))
+        )
+        test_assert("all assets are installed to /www", origin_files, target_files)
+    finally:
+        delete_path(www_root)
+
+
 def test_registration():
     test_assert(
         "simple route registration",
@@ -297,6 +313,7 @@ async def test_main():
     await test_chunked_transfer_encoding()
     await test_fs_access_control()
     await test_keepalive()
+    test_www_install()
 
 
 asyncio.run(test_main())