diff --git a/.gitignore b/.gitignore index 7edd53f..384ec00 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,8 @@ venv/ __pycache__ pyrobusta.env +pyrobusta.passwd +pyrobusta.roles build/ runtime/ runtime-test/ diff --git a/Makefile b/Makefile index 168d012..1981023 100644 --- a/Makefile +++ b/Makefile @@ -100,7 +100,10 @@ deploy-config: @echo "Uploading pyrobusta.env" @mpremote $(DEVICE) soft-reset @if [ -f pyrobusta.env ]; then mpremote $(DEVICE) cp pyrobusta.env :pyrobusta.env; fi + @if [ -f pyrobusta.roles ]; then mpremote $(DEVICE) cp pyrobusta.roles :pyrobusta.roles; fi + @if [ -f pyrobusta.passwd ]; then mpremote $(DEVICE) cp pyrobusta.passwd :pyrobusta.passwd; fi @mpremote $(DEVICE) reset + @sleep 5 # ----------------------------- @@ -191,6 +194,8 @@ stage-app: @cp $(TLS_DIR)/key.der $(RUNTIME_DIR)/ @if [ -f pyrobusta.env ]; then cp pyrobusta.env $(RUNTIME_DIR)/; fi + @if [ -f pyrobusta.roles ]; then cp pyrobusta.roles $(RUNTIME_DIR)/; fi + @if [ -f pyrobusta.passwd ]; then cp pyrobusta.passwd $(RUNTIME_DIR)/; fi @echo "http_port=8080" >> $(RUNTIME_DIR)/pyrobusta.env @echo "https_port=4443" >> $(RUNTIME_DIR)/pyrobusta.env @@ -212,9 +217,7 @@ deploy-app: mpremote $(DEVICE) cp $(APP_DIR)/boot.py :boot.py mpremote $(DEVICE) cp $(APP_DIR)/app.py :app.py - @echo "Uploading pyrobusta.env" - @if [ -f pyrobusta.env ]; then mpremote $(DEVICE) cp pyrobusta.env :pyrobusta.env; fi - @mpremote $(DEVICE) reset + $(MAKE) deploy-config @echo "\e[32m$(APP_DIR) app is successfully deployed, \n"\ "run 'make DEVICE=$(DEVICE) run-device' to restart the device and check the output.\e[0m" @@ -299,6 +302,7 @@ test-unix: stage-test tls-cert .PHONY: test-device test-device: stage-test #clean-device upload @mpremote $(DEVICE) soft-reset + @mpremote $(DEVICE) cp $(TEST_RUNTIME)/env_utils.py :/env_utils.py @cd $(TEST_RUNTIME); \ for test in test_*.py; do \ echo "\n==================================="; \ diff --git a/README.md b/README.md index eb36c1a..112a146 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ PyRobusta is a memory-conscious HTTP/1.1 server library built for embedded devic - Built-in API for uploading, downloading, and deleting files stored on the server - Persistent connection handling via the `Connection: keep-alive` header - HTTP/1.0 and HTTP/1.1 support +- Basic authentication and RBAC authorization - TLS support ## Design Principles diff --git a/docs/application_development/configuration.md b/docs/application_development/configuration.md index 6eb8582..86a0e4c 100644 --- a/docs/application_development/configuration.md +++ b/docs/application_development/configuration.md @@ -51,9 +51,12 @@ $ mpremote a0 cp pyrobusta.env :/pyrobusta.env | `http_mem_cap` | Fraction of available heap memory reserved for stream buffers. Valid range: (0, 1]. | 0.1 | | `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 | | `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 | -| `log_level` | Logging level. Can be one of: `warning`, `info`, `debug`. | info | +| `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` | ## Configuration API diff --git a/docs/application_development/index.md b/docs/application_development/index.md index f97dcd5..2784d8e 100644 --- a/docs/application_development/index.md +++ b/docs/application_development/index.md @@ -17,6 +17,7 @@ Then continue with the remaining guides in the order listed below. * [Response Processing](response.md) * [Static Content](static_content.md) * [File Server API](file_server.md) +* [Authentication & Security](security.md) * [Source Code](https://github.com/szeka9/PyRobusta) --- diff --git a/docs/application_development/request.md b/docs/application_development/request.md index 11a986d..9552840 100644 --- a/docs/application_development/request.md +++ b/docs/application_development/request.md @@ -155,7 +155,8 @@ Unlike regular request bodies, multipart parts are parsed by the server before b to the application. Preprocessed parts consist of headers (dictionary) and the raw part body (bytes), passed as a tuple. -**Multipart state tracking** +### Multipart State Tracking + The HTTP context exposes the boolean attributes `mp_is_first` and `mp_is_last` diff --git a/docs/application_development/security.md b/docs/application_development/security.md index 5f7038d..be512c9 100644 --- a/docs/application_development/security.md +++ b/docs/application_development/security.md @@ -2,26 +2,199 @@ [← Back](index.md) -This page is work in progress. Refer to the [main page](index.md) for available documentation. +This page documents the security features provided by PyRobusta, +including HTTP Basic Authentication, role-based authorization, and TLS. + +Note: Authentication verifies the client's identity. Authorization determines +whether the authenticated user is permitted to access a resource. --- ## Table of Contents * [Authentication & Security](#authentication-security) - + [API Key Authentication](#api-key-authentication) + [Basic Authentication](#basic-authentication) + + [Authorization](#authorization) + [HTTPS / TLS](#https-tls) + [Certificate Installation](#certificate-installation) --- -## API Key Authentication - ## Basic Authentication +PyRobusta supports HTTP Basic Authentication with per-user credentials +stored on the device. Passwords are stored as SHA-256 hashes. +Basic authentication is disabled by default. It can be enabled by setting +`http_auth=basic` in [pyrobusta.env](configuration.md). + +During initialization, PyRobusta reads `pyrobusta.passwd` by default +from the server's working directory. Each entry specifies a username, +a password hash, and one or more role names. Set `passwd_file` in the configuration +to specify an alternative user credentials file to load. For more information, +see [Server Configuration](./configuration.md). + +PyRobusta defines a single realm (`realm="Device"`) to authenticate against. +The server response includes the `WWW-Authenticate: Basic realm="Device"` header +when a request does not contain valid credentials. + +### pyrobusta.passwd + +`pyrobusta.passwd` utilizes a passwd-like format, specifying one user per line, +with each element of the user data separated by `:`. The expected format +of a user entry is `::`. + +``` +# Example +# File: /pyrobusta.passwd +szeka9:9b085649ac73a164314f00ec70010ec37d6c663cf71e9672a0600d6609b3ae12:api_admin +user-1:c0418ce1f43b9013c6871d4ac4941de6cd83a3fac29677785b250b86c88320a0:api_user,app_viewer +[...] +user-n:a8105c3c8b75556a9099b8dcab9cc13362d5a6f9fa5888ce39efc57961deb519:api_user,app_maintainer +``` + +When adding new users, follow the below steps to calculate the password hash: + +```python3 +import hashlib + +password_raw = "" +password_hash = hashlib.sha256(password_raw.encode()).digest().hex() +print(password_hash) # paste the output into pyrobusta.passwd + +b9fa35baa8069f3dabe214aed3525da0824efefdbce5ad83dfe4e7f48f8ce15f +``` + +## Authorization + +When HTTP Basic Authentication is enabled, PyRobusta provides role-based access control (RBAC). +Resource permissions are determined by the authenticated user's assigned roles. +Each user can be assigned one or more roles, configured in `pyrobusta.passwd`. + +While `pyrobusta.passwd` assigns roles to users, `pyrobusta.roles` maps those roles to +HTTP methods and server resources. Authorization follows a least-privilege model. +Users have no permissions unless explicitly granted by one or more authorization rules. +Each matching rule grants additional permissions. Requests that do not match an authorization +rule receive HTTP 403 Forbidden. + +Set `roles_file` in the configuration to specify an alternative role configuration file to load. +For more information, see [Server Configuration](./configuration.md). + +### pyrobusta.roles + +`pyrobusta.roles` consists of one or more authorization blocks. Each block defines one or +more path patterns, followed by one or more authorization rules. Every authorization rule +applies to every path pattern declared within the same block. + +``` + + +... + : ,,... + : ,,... + ... + + + ... +``` + +Authorization rules are selected based on pattern specificity rather than their +order in `pyrobusta.roles`. Path patterns support the following forms, ordered +from least to most specific. Each request is matched against the most specific +pattern, and only the roles associated with the selected pattern are considered +during authorization. + +``` +/** # Match any path +/* # Match direct children of / +/app # Exact match +/app/** # Match any descendant of /app +/app/* # Match direct children of /app +/app/*/resources # Match a single wildcard segment +/app/endpoint/resources # Exact match +``` + +**Note**: recursive globs (`**`) are only supported as the final path segment. +Allowing `**` in intermediate segments would introduce ambiguous matches +and is therefore not supported. + +``` +# File: /pyrobusta.roles + +# --------------------------------------------------------- +# Example: public resources (/, /index.html, /styles.css, ...) +/ +/* + GET,HEAD: * + +# --------------------------------------------------------- +# Example: apply rule to parent and child resources +/app/resource +/app/resource/* + GET: resource_viewer + POST,PUT,DELETE: resource_maintainer + +# --------------------------------------------------------- +# Example: apply rule to paths of any length with ** +/app/api/** + GET: api_viewer + POST,PUT,DELETE: api_maintainer + +# --------------------------------------------------------- +# Example: require api_admin access for all HTTP methods +/app/api/management + *: api_admin + +# --------------------------------------------------------- +# Example: restrict all access to a resource +/app/secret + *: + +# --------------------------------------------------------- +# Example: assign multiple roles +/app/logs + GET: api_admin,api_maintainer +``` + +**Note**: The special role `*` grants public access regardless of the authenticated user's roles. +Rules that specify `*` do not require authentication. + +## Authentication & Authorization Flow + +```mermaid +flowchart TD + A[Request] --> B[Resolve requested resource] + B --> C[Determine security policy for resource] + + C --> D[Public] + C --> E[Protected] + + D --> F[Ignore auth requirement] + F --> G[Process request] + + E --> H{Is 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?} + + M -->|No| N[403] + M -->|Yes| G +``` + +| Condition | Response | +| --- | --- | +| Missing credentials | 401 Unauthorized | +| Invalid credentials | 401 Unauthorized | +| Authenticated but insufficient permissions | 403 Forbidden | + ## HTTPS / TLS + + ## Certificate Installation --- diff --git a/src/pyrobusta/protocol/http.py b/src/pyrobusta/protocol/http.py index f58a1ad..225550f 100644 --- a/src/pyrobusta/protocol/http.py +++ b/src/pyrobusta/protocol/http.py @@ -12,6 +12,7 @@ CONF_HTTP_MULTIPART, CONF_HTTP_FILES_API, CONF_HTTP_SERVED_PATHS, + CONF_HTTP_AUTH, ) from ..utils import logging, helpers from ..stream.buffer import BufferFullError @@ -49,6 +50,7 @@ class HttpEngine: - http_files_api: serve files at the /files API, with support for uploads, removal and directory listing - http_multipart: support for multipart requests/responses + - http_auth: authenticate and authorize users """ __slots__ = ( @@ -79,6 +81,8 @@ class HttpEngine: b"204 No Content", 400, b"400 Bad Request", + 401, + b"401 Unauthorized", 403, b"403 Forbidden", 404, @@ -692,6 +696,11 @@ def _parse_headers_st(self, rx): if self.version == b"HTTP/1.1" and "host" not in self.headers: raise InvalidHeaders() rx.consume(blank_idx + 4) + self.state = self._handle_auth_st + + def _handle_auth_st(self, _): + # This a placeholder for authentication & authorization + # purposes to be overridden by extensions. self.state = self._route_request_st def _route_request_st(self, _): @@ -904,3 +913,8 @@ def enable_optional_features(): from pyrobusta.protocol import http_file_server http_file_server.apply_patches() + + if get_config(CONF_HTTP_AUTH) == "basic": + from pyrobusta.protocol import http_basic_auth + + http_basic_auth.apply_patches() diff --git a/src/pyrobusta/protocol/http_basic_auth.py b/src/pyrobusta/protocol/http_basic_auth.py new file mode 100644 index 0000000..21460fb --- /dev/null +++ b/src/pyrobusta/protocol/http_basic_auth.py @@ -0,0 +1,346 @@ +""" +Module for HTTP Basic Authentication. + +This module overrides the auth placeholder HttpEngine._handle_auth_st(), +and applies the basic authentication scheme, with hash-based password storage, +and role-based authorization. +""" + +# pylint: disable=W0212,R0401 + +import hashlib +import gc +import binascii + +from pyrobusta.protocol import http +from pyrobusta.utils.patch import add_method +from pyrobusta.utils.helpers import iterate_segments +from pyrobusta.utils.config import get_config, CONF_PASSWD_FILE, CONF_ROLES_FILE +from pyrobusta.utils.logging import warning + +_MAX_ROLES = 32 +_NO_POLICY = 2**_MAX_ROLES +_ALL_ROLES = 2**_MAX_ROLES - 1 + +_ROLE_INDEX = {} +_USERS = {} # {"user": [b"", b""]} +_ATTR_TREE = None # root node: / + +_DUMMY_HASH = hashlib.sha256(b"invalid-user").digest() + + +class AttributeNode: + """ + Tree-based data structure for authorization. + The tree represents predefined authorization masks defined + for different segments of URL paths. + """ + + __slots__ = ("name", "children", "attributes") + + def __init__(self, name: str): + self.name = name + self.children = [] + self.attributes = None + + def add_child(self, node): + """ + Add a child node to the current node. + """ + self.children.append(node) + + def iter_tree(self, path: str, glob: bool = False): + """ + Generator for iterating over the nodes of the tree + on a predefined path. + """ + current_node = self + for segment in iterate_segments(path, "/"): + segment = segment.lower() + if not segment: + continue + child_node = None + while child_node is None: + for child in current_node.children: + if child.name == segment: + child_node = child + break + if glob and child_node is None: + for child in current_node.children: + if child.name == "*": + child_node = child + break + if child_node is None: + # Yield (None, segment) until the caller inserts the + # missing child, or handles this condition in another way. + # Once the child exists, resume traversal automatically. + yield None, segment + yield child_node, segment + current_node = child_node + + def insert_path(self, path: str, attributes: dict): + """ + Introduce missing nodes in the tree based on + a specific path with attributes defined for the + last node corresponding to the trailing path segment. + """ + if not path.startswith("/") or path.find("/**/") != -1: + raise ValueError() + current_node = self + for iter_node, segment in self.iter_tree(path, glob=False): + if not iter_node: + n = AttributeNode(segment) + current_node.add_child(n) + else: + current_node = iter_node + current_node.attributes = attributes.copy() + + def get_attributes(self, path: str): + """ + Retrieve the attributes of a node, corresponding + to the last segment of a path. Use glob-based rules + if nodes of trailing path segments are missing. + - when a single path segment is missing: resolve attributes from '*' node + - when multiple trailing path segments are missing: resolve attributes from '**' node + """ + current_node = self + + # Calculate number of segments for globbing + num_segments = 0 + prev = "/" + for c in path: + if prev == "/" and c != "/": + num_segments += 1 + prev = c + + # Iterate nodes + parent_glob = ( + None # Track parent attributes corresponding to recursive globs (**) + ) + num_nodes = 0 + + for iter_node, _ in self.iter_tree(path, glob=True): + num_nodes += 1 + glob = None + + for child in current_node.children: + if num_nodes == num_segments and child.name == "*": + glob = child + if child.name == "**": + parent_glob = child + + current_node = iter_node + if not iter_node: + break + + if not current_node: + if glob: + return glob.attributes + if parent_glob: + return parent_glob.attributes + return + return current_node.attributes + + +def index_roles(roles: str): + """ + Create a common index for roles defined in user + definitions and role definitions, and assign a binary + mask to each role, used for authorization. + """ + role_mask = 0b0 + for role in iterate_segments(roles, ","): + role = role.strip().lower() + if not role: + continue + if role not in _ROLE_INDEX and role != "*": + if len(_ROLE_INDEX) == _MAX_ROLES: + raise ValueError() + _ROLE_INDEX[role] = len(_ROLE_INDEX) + if role == "*": + role_mask = _NO_POLICY + else: + role_mask |= 1 << _ROLE_INDEX[role] + return role_mask + + +def _load_users(): + passwd_file = get_config(CONF_PASSWD_FILE) + try: + with open(passwd_file, encoding="utf-8") as users: + for line in users: + comment_idx = line.find("#") + line = line[:comment_idx] if comment_idx != -1 else line + if not line: + continue + if not line.count(":") == 2: + raise ValueError() + user_sep = line.find(":") + password_sep = line.find(":", user_sep + 1) + user = line[:user_sep].strip().lower() + password_hash = line[user_sep + 1 : password_sep].strip() + roles = line[password_sep + 1 :] + role_mask = index_roles(roles) + if not user or not password_hash: + raise ValueError() + if user in _USERS: + raise ValueError() + _USERS[user] = [bytes.fromhex(password_hash), role_mask] + except OSError: + warning("Unable to open: " + passwd_file) + + +def _load_roles(): + roles_file = get_config(CONF_ROLES_FILE) + try: + with open(roles_file, encoding="utf-8") as roles: + paths = [] + attributes = {} + for line in roles: + comment_idx = line.find("#") + line = line[:comment_idx].strip() if comment_idx != -1 else line.strip() + + # Parse URL path + if line.startswith("/"): + if paths and attributes: + for path in paths: + _ATTR_TREE.insert_path(path, attributes) + paths = [line] + attributes = {} + else: + paths.append(line) + + # Parse attributes + elif line: + if not paths: + raise ValueError() + sep = line.find(":") + if sep in (0, -1): + raise ValueError() + role_mask = index_roles(line[sep + 1 :]) + for attr in iterate_segments(line[0:sep].strip(), ","): + if attr in attributes: + raise ValueError() + if attr: + attributes[attr] = role_mask + for path in paths: + _ATTR_TREE.insert_path(path, attributes) + except OSError: + warning("Unable to open: " + roles_file) + + +def _handle_auth_st(self, _): + # Determine security policy + is_public = False + method = self.method.decode("ascii") + url = self.url.decode("ascii") + + policy = _ATTR_TREE.get_attributes(url) + if not policy: + self.state = self._handle_auth_header_st + return + + if method not in policy: + if policy.get("*") == _NO_POLICY: + is_public = True + elif policy[method] == _NO_POLICY: + is_public = True + + if is_public: + self.state = self._route_request_st + else: + 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"' + + # 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 + + # Decoding + user_data = auth_header[5:].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 + + # Authentication + user_sep = user_data.find(":") + if user_sep < 0: + self.set_response_header(www_auth_header, www_auth_method) + self.terminate(401) + return + + username = user_data[:user_sep].lower() + user_info = _USERS.get(username) + stored_hash = user_info[0] if user_info else _DUMMY_HASH + password_hash = hashlib.sha256( + user_data[user_sep + 1 :].strip().encode("ascii") + ).digest() + + hash_ok = compare_digest(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) + self.terminate(401) + return + + # Authorization + policy = _ATTR_TREE.get_attributes(url) + + if not policy: + allowed_roles = 0 + elif method not in policy: + allowed_roles = policy.get("*", 0) + else: + allowed_roles = policy[method] + + if (allowed_roles & user_info[1]) == 0: + self.terminate(403) + return + + self.state = self._route_request_st + + +def apply_patches(): + """ + Apply patches to class attributes for HTTP basic authentication. + """ + global _ATTR_TREE # pylint: disable=W0603 + try: + add_method(http.HttpEngine, _handle_auth_st) + add_method(http.HttpEngine, _handle_auth_header_st) + + _ATTR_TREE = AttributeNode("") + if _USERS: + _USERS.clear() + + _load_users() + _load_roles() + finally: + # Clean up temporary data structures + _ROLE_INDEX.clear() + gc.collect() diff --git a/src/pyrobusta/utils/config.py b/src/pyrobusta/utils/config.py index 1acb070..be194aa 100644 --- a/src/pyrobusta/utils/config.py +++ b/src/pyrobusta/utils/config.py @@ -29,9 +29,12 @@ def const(n): # pylint: disable=C0116 CONF_HTTP_MEM_CAP = const(5) CONF_HTTP_SERVED_PATHS = const(6) CONF_HTTP_FILES_API = const(7) -CONF_SOCKET_MAX_CON = const(8) -CONF_TLS = const(9) -CONF_LOG_LEVEL = const(10) +CONF_HTTP_AUTH = const(8) +CONF_SOCKET_MAX_CON = const(9) +CONF_TLS = const(10) +CONF_LOG_LEVEL = const(11) +CONF_PASSWD_FILE = const(12) +CONF_ROLES_FILE = const(13) # ------------------- # Configuration state @@ -51,21 +54,28 @@ def const(n): # pylint: disable=C0116 CONF_HTTP_MEM_CAP, 0.1, CONF_HTTP_SERVED_PATHS, - ["/www", "/lib/pyrobusta"], + [normalize_path("/www"), normalize_path("/lib/pyrobusta")], CONF_HTTP_FILES_API, False, + CONF_HTTP_AUTH, + None, 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"), ] # -------------- # Public helpers # -------------- +# pylint: disable=R0911 def parse_config(key, value): """ Normalize a configuration value depending on the key. @@ -78,9 +88,11 @@ 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 not in (CONF_WIFI_SSID, CONF_WIFI_PASSWORD): - return value.lower() - return value + if key in (CONF_PASSWD_FILE, CONF_ROLES_FILE): + return normalize_path(value) + if key in (CONF_WIFI_SSID, CONF_WIFI_PASSWORD): + return value + return value.lower() def read_config(config=CONFIG_LOCATION): @@ -122,7 +134,7 @@ def get_config(key): or the value is set to None. """ global _CONFIG_LOADED # pylint: disable=W0603 - if _CONFIG_CACHE[2 * key + 1] is None or not _CONFIG_LOADED: + if not _CONFIG_LOADED: read_config() _CONFIG_LOADED = True return _CONFIG_CACHE[2 * key + 1] diff --git a/src/pyrobusta/utils/helpers.py b/src/pyrobusta/utils/helpers.py index 11fcf90..7c6432d 100644 --- a/src/pyrobusta/utils/helpers.py +++ b/src/pyrobusta/utils/helpers.py @@ -77,3 +77,22 @@ def is_path_segment_valid(filename: str): ): return False return True + + +def iterate_segments(data: str, delimiter: str): + """ + Generator for iterating over each segment of a string, + separated by an arbitrary delimiter. The generator + does not ignore empty segments (including leading and + trailing segments). + """ + start = 0 + end = 0 + while start < len(data): + end = data.find(delimiter, start) + if end == -1: + end = len(data) + yield data[start:end].strip() + start = end + 1 + if data.endswith(delimiter): + yield "" diff --git a/tests/.pylintrc b/tests/.pylintrc index 533feb6..c8dbc83 100644 --- a/tests/.pylintrc +++ b/tests/.pylintrc @@ -1,5 +1,6 @@ [MAIN] -disable=W0212, +disable=W0201, + W0212, C0114, C0115, C0116, @@ -7,4 +8,8 @@ disable=W0212, R0902, R0801 -init-hook='import os, sys; tests = os.path.join(os.getcwd(), "tests"); [sys.path.append(d) for d, _, _ in os.walk(tests)];' \ No newline at end of file +init-hook='import os, sys; tests = os.path.join(os.getcwd(), "tests"); [sys.path.append(d) for d, _, _ in os.walk(tests)];' + +[DESIGN] +max-args=7 +max-positional-arguments=7 diff --git a/tests/functional/env_utils.py b/tests/functional/env_utils.py new file mode 100644 index 0000000..b7aebdb --- /dev/null +++ b/tests/functional/env_utils.py @@ -0,0 +1,119 @@ +import asyncio +import ssl +import gc + +from os import mkdir, listdir, remove, rmdir + +from pyrobusta.server import http_server +from pyrobusta.protocol.http import enable_optional_features +from pyrobusta.utils.config import ( + CONF_TLS, + CONF_LOG_LEVEL, + CONF_HTTP_MULTIPART, + CONF_HTTP_FILES_API, + CONF_HTTP_SERVED_PATHS, + CONF_HTTP_AUTH, + _CONFIG_CACHE, + parse_config, +) + + +def garbage_collect(coroutine): + async def decorated(*args, **kwargs): + gc.collect() + await coroutine(*args, **kwargs) + gc.collect() + + return decorated + + +def test_assert(name, actual, expected): + print(f"Test {name}: ", end="") + if actual == expected: + print("OK") + else: + print("Fail") + raise AssertionError(f"{actual} != {expected}") + + +async def start_server(): + """ + Start an HTTP server as a background task. + """ + server = http_server.HttpServer() + await server.start_socket_server() + await asyncio.sleep_ms(100) + return server + + +async def send_request(request, tls=False): + port = ( + http_server.HttpServer.LISTEN_PORT_HTTPS + if tls + else http_server.HttpServer.LISTEN_PORT_HTTP + ) + + ctx = None + if tls: + # Disable certificate verification due to self-signed cert + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.verify_mode = ssl.CERT_NONE + + reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) + writer.write(request) + await writer.drain() + + to_read = True + response = b"" + while to_read: + response_part = await reader.read(1024) + response += response_part + to_read = len(response_part) + writer.close() + return response + + +def fmkdir(path: str): + try: + mkdir(path) + except OSError: + pass + + +def delete_path(path): + for name in listdir(path): + if path == "/": + full = "/" + name + else: + full = path + "/" + name + + try: + remove(full) + except OSError: + delete_path(full) + try: + rmdir(full) + except OSError: + pass + + +def setup_config( + tls_enabled=False, + files_api_enabled=False, + http_multipart_enabled=False, + served_paths="", + http_auth="", +): + 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( + 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 + + enable_optional_features() diff --git a/tests/functional/test_http.py b/tests/functional/test_http.py index f459dad..92905b8 100644 --- a/tests/functional/test_http.py +++ b/tests/functional/test_http.py @@ -1,99 +1,20 @@ import asyncio -import ssl import json -import gc -from os import mkdir, remove, rmdir, listdir +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, + fmkdir, + delete_path, +) from pyrobusta.server import http_server from pyrobusta.protocol.http import HttpEngine -from pyrobusta.utils.config import ( - CONF_HTTP_SERVED_PATHS, - CONF_HTTP_FILES_API, - CONF_TLS, - CONF_LOG_LEVEL, - _CONFIG_CACHE, - parse_config, -) -from pyrobusta.utils.helpers import normalize_path - -################################################# -# Test helpers -################################################# - - -def garbage_collect(coroutine): - async def decorated(*args, **kwargs): - gc.collect() - await coroutine(*args, **kwargs) - gc.collect() - - return decorated - - -def test_assert(name, actual, expected): - print(f"Test {name}: ", end="") - if actual == expected: - print("OK") - else: - print("Fail") - raise AssertionError(f"{actual} != {expected}") - - -async def send_request(request, tls=False): - port = ( - http_server.HttpServer.LISTEN_PORT_HTTPS - if tls - else http_server.HttpServer.LISTEN_PORT_HTTP - ) - - ctx = None - if tls: - # Disable certificate verification due to self-signed cert - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.verify_mode = ssl.CERT_NONE - - reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) - writer.write(request) - await writer.drain() - - to_read = True - response = b"" - while to_read: - response_part = await reader.read(1024) - response += response_part - to_read = len(response_part) - writer.close() - return response - -def fmkdir(path: str): - try: - mkdir(path) - except OSError: - pass - - -def delete_path(path): - for name in listdir(path): - if path == "/": - full = "/" + name - else: - full = path + "/" + name - - try: - remove(full) - except OSError: - delete_path(full) - try: - rmdir(full) - except OSError: - pass - - -################################################# -# Test driver -################################################# +from pyrobusta.utils.helpers import normalize_path @HttpEngine.route("/test/simple", "GET") @@ -121,16 +42,6 @@ def chunked_handler(http_ctx, chunk): recv_chunks.append(chunk.decode("utf8")) -async def start_server(): - """ - Start an HTTP server as a background task - """ - server = http_server.HttpServer() - await server.start_socket_server() - await asyncio.sleep_ms(100) - return server - - @garbage_collect async def test_simple_response(tls_enabled): setup_config(tls_enabled=tls_enabled) @@ -366,23 +277,6 @@ async def test_keepalive(): await server.terminate() -################################################# -# Test methods -################################################# - - -def setup_config(tls_enabled=False, served_paths="", files_api_enabled=False): - 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( - CONF_HTTP_SERVED_PATHS, served_paths - ) - _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = files_api_enabled - - def test_registration(): test_assert( "simple route registration", @@ -397,15 +291,14 @@ def test_registration(): ) -def test_main(): +async def test_main(): test_registration() - asyncio.run(test_simple_response(tls_enabled=False)) - asyncio.run(test_simple_response(tls_enabled=True)) - - asyncio.run(test_server_busy()) - asyncio.run(test_chunked_transfer_encoding()) - asyncio.run(test_fs_access_control()) - asyncio.run(test_keepalive()) + await test_simple_response(tls_enabled=False) + await test_simple_response(tls_enabled=True) + await test_server_busy() + await test_chunked_transfer_encoding() + await test_fs_access_control() + await test_keepalive() -test_main() +asyncio.run(test_main()) diff --git a/tests/functional/test_http_basic_auth.py b/tests/functional/test_http_basic_auth.py new file mode 100644 index 0000000..7cba210 --- /dev/null +++ b/tests/functional/test_http_basic_auth.py @@ -0,0 +1,127 @@ +import asyncio + +from pyrobusta.protocol import http_basic_auth +from pyrobusta.protocol.http import HttpEngine +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, +) + + +@HttpEngine.route("/test/auth", "GET") +def auth_handler(http_ctx, _): + return "text/plain", "OK" + + +@garbage_collect +async def test_missing_auth_header(): + setup_config(http_auth="basic") + server = await start_server() + + # Test: unauthenticated & unauthorized + plain_response = await send_request( + b"GET /test/auth HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Length: 0\r\n" + b"Connection: close\r\n\r\n" + ) + + test_assert( + f"request rejected with 401 Unauthorized", + b"401 Unauthorized" in plain_response, + True, + ) + + await server.terminate() + + +@garbage_collect +async def test_missing_role(): + setup_config(http_auth="basic") + server = await start_server() + + # Test: authenticated & unauthorized + plain_response = await send_request( + b"GET /test/auth HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Length: 0\r\n" + b"Authorization: Basic dXNlcjE6cGFzc3dvcmQx\r\n" + b"Connection: close\r\n\r\n" + ) + + test_assert( + f"request rejected with 403 Forbidden", + b"403 Forbidden" in plain_response, + True, + ) + + await server.terminate() + + +@garbage_collect +async def test_user_authorized(): + setup_config(http_auth="basic") + server = await start_server() + + # Test: authenticated & authorized + plain_response = await send_request( + b"GET /test/auth HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Length: 0\r\n" + b"Authorization: Basic dXNlcjI6cGFzc3dvcmQy\r\n" + b"Connection: close\r\n\r\n" + ) + + test_assert( + f"request accepted with 200 OK", + b"200 OK" in plain_response, + True, + ) + + await server.terminate() + + +def setup_auth(): + with open("pyrobusta.passwd", "w") as users: + users.write( + "user1:0b14d501a594442a01c6859541bcb3e8164d183d32937b851835442f69d5c94e:role1\n" + ) + users.write( + "user2:6cf615d5bcaac778352a8f1f3360d23f02f34ec182e259897fd6ce485d7870d4:role2\n" + ) + + with open("pyrobusta.roles", "w") as users: + users.write("/test/auth\n*:role2\n") + + +def test_registration(): + test_assert( + "auth route registration", + auth_handler, + HttpEngine._get_handler(b"/test/auth", b"GET"), + ) + + +def test_auth_patches(): + setup_auth() + setup_config(http_auth="basic") + test_assert( + "auth state machine patches", + http_basic_auth._handle_auth_st, + HttpEngine._handle_auth_st, + ) + + +async def test_main(): + test_registration() + test_auth_patches() + + await test_missing_auth_header() + await test_missing_role() + await test_user_authorized() + + +asyncio.run(test_main()) diff --git a/tests/functional/test_http_file_server.py b/tests/functional/test_http_file_server.py index 618d860..3439c67 100644 --- a/tests/functional/test_http_file_server.py +++ b/tests/functional/test_http_file_server.py @@ -1,132 +1,24 @@ import asyncio import json -import json -import ssl -import gc -from os import mkdir, listdir, remove, rmdir, stat +from os import stat -from pyrobusta.server import http_server -from pyrobusta.protocol.http import ( - CONF_HTTP_SERVED_PATHS, - enable_optional_features, - stat, -) -from pyrobusta.utils.config import ( - CONF_TLS, - CONF_LOG_LEVEL, - CONF_HTTP_MULTIPART, - CONF_HTTP_FILES_API, - _CONFIG_CACHE, - normalize_path, - parse_config, +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, + fmkdir, + delete_path, ) -################################################# -# Test helpers -################################################# - - -def garbage_collect(coroutine): - async def decorated(*args, **kwargs): - gc.collect() - await coroutine(*args, **kwargs) - gc.collect() - - return decorated - - -def test_assert(name, actual, expected): - print(f"Test {name}: ", end="") - if actual == expected: - print("OK") - else: - print("Fail") - raise AssertionError(f"{actual} != {expected}") - - -async def send_request(request, tls=False): - port = ( - http_server.HttpServer.LISTEN_PORT_HTTPS - if tls - else http_server.HttpServer.LISTEN_PORT_HTTP - ) - - ctx = None - if tls: - # Disable certificate verification due to self-signed cert - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.verify_mode = ssl.CERT_NONE - - reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) - writer.write(request) - await writer.drain() - - to_read = True - response = b"" - while to_read: - response_part = await reader.read(1024) - response += response_part - to_read = len(response_part) - writer.close() - return response - - -def multipart_response(num_responses): - i = 0 - - def response_generator(): - nonlocal i - i += 1 - if i > num_responses: - return None - return "text/plain", b"Response %s" % i - - return response_generator - - -def fmkdir(path: str): - try: - mkdir(path) - except OSError: - pass - - -def delete_path(path): - for name in listdir(path): - if path == "/": - full = "/" + name - else: - full = path + "/" + name - - try: - remove(full) - except OSError: - delete_path(full) - try: - rmdir(full) - except OSError: - pass - - -################################################# -# Test driver -################################################# - - -async def start_server(): - """ - Start an HTTP server as a background task. - """ - server = http_server.HttpServer() - await server.start_socket_server() - await asyncio.sleep_ms(100) - return server +from pyrobusta.utils.config import normalize_path @garbage_collect async def test_fs_path_traversal(): - setup_config(served_paths="/test") + setup_config(files_api_enabled=True, served_paths="/test") server = await start_server() test_root = normalize_path("/test") styles_dir = normalize_path("/test/style") @@ -145,6 +37,7 @@ async def test_fs_path_traversal(): # Test case response = await send_request( b"GET /files/test HTTP/1.1\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" b"Host: localhost\r\n\r\n" ) @@ -187,7 +80,7 @@ async def test_fs_path_traversal(): @garbage_collect async def test_fs_access_control(): - setup_config(served_paths="/test/allowed") + setup_config(files_api_enabled=True, served_paths="/test/allowed") server = await start_server() test_root = normalize_path("/test") @@ -211,6 +104,7 @@ async def test_fs_access_control(): # Case #1: /test/allowed/index.html response = await send_request( b"GET /files/test/allowed/index.html HTTP/1.1\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" b"Host: localhost\r\n\r\n" ) @@ -225,6 +119,7 @@ async def test_fs_access_control(): # Case #2: /test/rejected/index.html response = await send_request( b"GET /files/test/rejected/index.html HTTP/1.1\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" b"Host: localhost\r\n\r\n" ) @@ -241,7 +136,7 @@ async def test_fs_access_control(): @garbage_collect async def test_bulk_file_upload(): - setup_config(http_multipart_enabled=True) + setup_config(files_api_enabled=True, http_multipart_enabled=True) server = await start_server() user_data = normalize_path("/www/user_data") @@ -298,7 +193,7 @@ async def test_bulk_file_upload(): @garbage_collect async def test_chunked_file_upload(): - setup_config() + setup_config(files_api_enabled=True) server = await start_server() user_data = normalize_path("/www/user_data") @@ -341,30 +236,11 @@ async def test_chunked_file_upload(): await server.terminate() -################################################# -# Test methods -################################################# - - -def setup_config(tls_enabled=False, http_multipart_enabled=False, served_paths=""): - 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( - CONF_HTTP_SERVED_PATHS, served_paths - ) - _CONFIG_CACHE[2 * CONF_HTTP_MULTIPART + 1] = http_multipart_enabled - _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = True - enable_optional_features() - - -def test_main(): - asyncio.run(test_fs_path_traversal()) - asyncio.run(test_fs_access_control()) - asyncio.run(test_bulk_file_upload()) - asyncio.run(test_chunked_file_upload()) +async def test_main(): + await test_fs_path_traversal() + await test_fs_access_control() + await test_bulk_file_upload() + await test_chunked_file_upload() -test_main() +asyncio.run(test_main()) diff --git a/tests/functional/test_http_multipart.py b/tests/functional/test_http_multipart.py index 6fde540..692a980 100644 --- a/tests/functional/test_http_multipart.py +++ b/tests/functional/test_http_multipart.py @@ -1,71 +1,15 @@ import asyncio -import ssl -import gc -from os import mkdir, listdir, remove, rmdir - -from pyrobusta.server import http_server -from pyrobusta.protocol import http_multipart -from pyrobusta.protocol.http import ( - HttpEngine, - enable_optional_features, +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, ) -from pyrobusta.utils.config import ( - CONF_TLS, - CONF_LOG_LEVEL, - CONF_HTTP_MULTIPART, - CONF_HTTP_FILES_API, - _CONFIG_CACHE, -) - -################################################# -# Test helpers -################################################# - - -def garbage_collect(coroutine): - async def decorated(*args, **kwargs): - gc.collect() - await coroutine(*args, **kwargs) - gc.collect() - - return decorated - -def test_assert(name, actual, expected): - print(f"Test {name}: ", end="") - if actual == expected: - print("OK") - else: - print("Fail") - raise AssertionError(f"{actual} != {expected}") - - -async def send_request(request, tls=False): - port = ( - http_server.HttpServer.LISTEN_PORT_HTTPS - if tls - else http_server.HttpServer.LISTEN_PORT_HTTP - ) - - ctx = None - if tls: - # Disable certificate verification due to self-signed cert - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.verify_mode = ssl.CERT_NONE - - reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) - writer.write(request) - await writer.drain() - - to_read = True - response = b"" - while to_read: - response_part = await reader.read(1024) - response += response_part - to_read = len(response_part) - writer.close() - return response +from pyrobusta.protocol import http_multipart +from pyrobusta.protocol.http import HttpEngine def multipart_response(num_responses): @@ -81,67 +25,28 @@ def response_generator(): return response_generator -def fmkdir(path: str): - try: - mkdir(path) - except OSError: - pass - - -def delete_path(path): - for name in listdir(path): - if path == "/": - full = "/" + name - else: - full = path + "/" + name - - try: - remove(full) - except OSError: - delete_path(full) - try: - rmdir(full) - except OSError: - pass - - -################################################# -# Test driver -################################################# - - @HttpEngine.route("/test/multipart", "GET") def multipart_handler(http_ctx, _): part_count = int(http_ctx.headers["x-part-count"]) return "multipart/form-data", multipart_response(part_count) -async def start_server(): - """ - Start an HTTP server as a background task. - """ - server = http_server.HttpServer() - await server.start_socket_server() - await asyncio.sleep_ms(100) - return server - - @garbage_collect -async def test_multipart_response(tls_enabled): - setup_config(tls_enabled=tls_enabled) +async def test_multipart_response(): + setup_config(http_multipart_enabled=True) server = await start_server() # Test: 1 part plain_response = await send_request( b"GET /test/multipart HTTP/1.1\r\n" b"Host: localhost\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" - b"X-Part-Count: 1\r\n\r\n", - tls_enabled, + b"X-Part-Count: 1\r\n\r\n" ) test_assert( - f"http{"s" if tls_enabled else ""} response contains 1 part", + f"multipart response contains 1 part", b"Response 1" in plain_response, True, ) @@ -150,12 +55,13 @@ async def test_multipart_response(tls_enabled): plain_response = await send_request( b"GET /test/multipart HTTP/1.1\r\n" b"Host: localhost\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" - b"X-Part-Count: 10\r\n\r\n", - tls_enabled, + b"X-Part-Count: 10\r\n\r\n" ) + test_assert( - f"http{"s" if tls_enabled else ""} response contains 10 parts", + f"multipart response contains 10 parts", [b"Response %s" % i in plain_response for i in range(1, 11)], [True] * 10, ) @@ -163,22 +69,6 @@ async def test_multipart_response(tls_enabled): await server.terminate() -################################################# -# Test methods -################################################# - - -def setup_config(tls_enabled=False, files_api_enabled=False): - 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_MULTIPART + 1] = True - _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = files_api_enabled - enable_optional_features() - - def test_registration(): test_assert( "multipart route registration", @@ -188,7 +78,7 @@ def test_registration(): def test_multipart_patches(): - setup_config() + setup_config(http_multipart_enabled=True) test_assert( "multipart state machine patches", http_multipart._start_multipart_parser_st, @@ -196,11 +86,10 @@ def test_multipart_patches(): ) -def test_main(): +async def test_main(): test_registration() test_multipart_patches() - asyncio.run(test_multipart_response(tls_enabled=False)) - asyncio.run(test_multipart_response(tls_enabled=True)) + await test_multipart_response() -test_main() +asyncio.run(test_main()) diff --git a/tests/unit/http_base.py b/tests/unit/http_base.py index 3f20c37..7756d50 100644 --- a/tests/unit/http_base.py +++ b/tests/unit/http_base.py @@ -11,6 +11,19 @@ class TestHttpBase(unittest.TestCase): Base class for HTTP file server module. """ + def patch_config_loader(self, config, config_module): + def open_side_effect(*args, **kwargs): + data = "\n".join(f"{k}={v}" for k, v in config.items()) + return mock_open(read_data=data)(*args, **kwargs) + + self.open_patcher = patch.object( + config_module, + "open", + side_effect=open_side_effect, + ) + self.open_patcher.start() + self.addCleanup(self.open_patcher.stop) + @classmethod def setUpClass(cls): cls.base_config = {} @@ -27,36 +40,43 @@ def setUp(self): self.cwd_patcher.start() self.addCleanup(self.cwd_patcher.stop) + # -------------------------------- + # Workspace, temporary directories + # -------------------------------- + if "tmp" not in os.listdir(self.cwd): + os.mkdir(self.cwd + "/tmp") + + if "passwd_file" in self.base_config: + self.passwd_file = self.cwd + self.base_config["passwd_file"] + with open(self.passwd_file, "w", encoding="utf-8"): + pass + if "roles_file" in self.base_config: + self.roles_file = self.cwd + self.base_config["roles_file"] + with open(self.roles_file, "w", encoding="utf-8"): + pass + # ------------------- # Patch config module # ------------------- self.config = dict(self.base_config) self.config_module = load_module("pyrobusta/utils/config.py") + self.patch_config_loader(self.config, self.config_module) + self.module_patcher = patch.dict( sys.modules, {"pyrobusta.utils.config": self.config_module}, ) + self.module_patcher.start() self.addCleanup(self.module_patcher.stop) - def open_side_effect(*args, **kwargs): - data = "\n".join(f"{k}={v}" for k, v in self.config.items()) - return mock_open(read_data=data)(*args, **kwargs) - - self.open_patcher = patch.object( - self.config_module, - "open", - side_effect=open_side_effect, - ) - self.open_patcher.start() - self.addCleanup(self.open_patcher.stop) - # ------------------------------------------------ # Load remaining modules, enable optional features # ------------------------------------------------ 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") + self.basic_auth_module = load_module("pyrobusta/protocol/http_basic_auth.py") self.fs_patcher = patch.object(self.fs_module, "setup_directories") self.fs_patcher.start() @@ -71,3 +91,12 @@ def open_side_effect(*args, **kwargs): buffer_module = load_module("pyrobusta/stream/buffer.py") self.rx = buffer_module.SlidingBuffer(bytearray(1024)) self.tx = buffer_module.SlidingBuffer(bytearray(1024)) + + def tearDown(self): + try: + if "passwd_file" in self.config: + os.remove(self.passwd_file) + if "roles_file" in self.config: + os.remove(self.roles_file) + finally: + super().tearDown() diff --git a/tests/unit/test_http.py b/tests/unit/test_http.py index 34b49db..8ab2193 100644 --- a/tests/unit/test_http.py +++ b/tests/unit/test_http.py @@ -172,7 +172,7 @@ def test_header_parsing_valid(self): self.engine.headers, ) self.assertEqual(self.rx.peek(), b"") - self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.state, self.engine._parse_headers_st) def test_header_parsing_incomplete_header(self): request = b"GET /index.html HTTP/1.1\r\nContent-Type\r\n\r\n" diff --git a/tests/unit/test_http_basic_auth.py b/tests/unit/test_http_basic_auth.py new file mode 100644 index 0000000..9778068 --- /dev/null +++ b/tests/unit/test_http_basic_auth.py @@ -0,0 +1,619 @@ +import os +import unittest +import hashlib +import base64 + +from http_base import TestHttpBase + + +class TestBasicAuthPolicyStateMachine(TestHttpBase): + """ + Tests for HTTP basic auth policy. + """ + + @classmethod + def setUpClass(cls): + cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } + + def test_basic_auth_public_resource(self): + self.basic_auth_module._ATTR_TREE.insert_path( + "/app/public", {"GET": self.basic_auth_module._NO_POLICY} + ) + self.engine.state = self.engine._handle_auth_st + self.engine.url = b"/app/public" + self.engine.method = b"GET" + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + + def test_basic_auth_private_resource_exact_rule(self): + self.basic_auth_module._ATTR_TREE.insert_path("/app/private", {"GET": 0b001}) + self.engine.state = self.engine._handle_auth_st + self.engine.url = b"/app/private" + self.engine.method = b"GET" + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._handle_auth_header_st) + + def test_basic_auth_private_resource_no_rule(self): + self.engine.state = self.engine._handle_auth_st + self.engine.url = b"/app/public" + self.engine.method = b"GET" + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._handle_auth_header_st) + + +class TestBasicAuthStateMachine(TestHttpBase): + """ + Tests for HTTP authentication & authorization. + """ + + @classmethod + def setUpClass(cls): + cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } + + 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] + + self.basic_auth_module._ATTR_TREE.insert_path( + "/app/private", {"GET": route_roles} + ) + self.engine.state = self.engine._handle_auth_header_st + self.engine.url = b"/app/private" + self.engine.method = b"GET" + if auth_header: + self.engine.headers["authorization"] = auth_header + + def test_basic_auth_successful(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.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.status_code, 401) + + def test_basic_auth_unknown_user(self): + self.prepare_auth( + user=None, + password=None, + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_user_nok(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "invalid:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_user_empty(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode(":".encode("ascii")).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_password_nok(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode("dummy-user:invalid".encode("ascii")).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_password_empty(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.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_role_nok(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b010, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 403) + + def test_basic_auth_header_missing(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="", + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_header_incomplete(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic ", + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_invalid_encoding(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic invalid-base64", + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_missing_colon(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.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_multiple_colon(self): + self.prepare_auth( + user="dummy-user", + password="dummy:password", + auth_header="Basic " + + base64.b64encode("dummy-user:dummy:password".encode("ascii")).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.status_code, 401) + + def test_basic_auth_scheme_case_sensitivity(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.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.status_code, 401) + + def test_basic_auth_invalid_scheme(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Bearer " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + +class TestBasicAuthPrefixTree(TestHttpBase): + """ + Tests for authorization based on prefix tree. + """ + + @classmethod + def setUpClass(cls): + cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } + + def setup_roles(self, roles: dict): + self.attr_tree = self.basic_auth_module.AttributeNode("") + for path, attributes in roles.items(): + self.attr_tree.insert_path(path, attributes) + + def test_attribute_retrieval_exact_match(self): + self.setup_roles({"/app/resource": {"GET": 0b001}}) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, {"GET": 0b001}) + + def test_attribute_retrieval_trailing_glob(self): + self.setup_roles({"/app/resource/*": {"GET": 0b001}}) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details/") + self.assertEqual(attributes, None) + + def test_attribute_retrieval_intermediate_glob(self): + self.setup_roles({"/app/resource/*/details": {"GET": 0b001}}) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details/") + self.assertEqual(attributes, {"GET": 0b001}) + + def test_attribute_retrieval_trailing_recursive_glob(self): + self.setup_roles( + { + "/app/resource/**": {"GET": 0b001}, + } + ) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details") + self.assertEqual(attributes, {"GET": 0b001}) + + def test_attribute_retrieval_glob_precedence(self): + self.setup_roles( + { + "/**": {"GET": 0}, + "/app": {"GET": 1}, + "/app/*": {"GET": 2}, + "/app/**": {"GET": 3}, + "/app/*/logs": {"GET": 4}, + "/app/faults/logs": {"GET": 5}, + "/app/endpoint": {"GET": 6}, + "/app/endpoint/*": {"GET": 7}, + "/app/endpoint/**": {"GET": 8}, + "/app/endpoint/*/logs": {"GET": 9}, + "/app/endpoint/runtime/logs": {"GET": 10}, + } + ) + attributes = self.attr_tree.get_attributes("/system/runtime/logs") + self.assertEqual(attributes, {"GET": 0}) + + attributes = self.attr_tree.get_attributes("/app") + self.assertEqual(attributes, {"GET": 1}) + + attributes = self.attr_tree.get_attributes("/app/api") + self.assertEqual(attributes, {"GET": 2}) + + attributes = self.attr_tree.get_attributes("/app/admin/status") + self.assertEqual(attributes, {"GET": 3}) + + attributes = self.attr_tree.get_attributes("/app/admin/logs") + self.assertEqual(attributes, {"GET": 4}) + + attributes = self.attr_tree.get_attributes("/app/faults/logs") + self.assertEqual(attributes, {"GET": 5}) + + attributes = self.attr_tree.get_attributes("/app/endpoint") + self.assertEqual(attributes, {"GET": 6}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/status") + self.assertEqual(attributes, {"GET": 7}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/resource/details") + self.assertEqual(attributes, {"GET": 8}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/resource/logs") + self.assertEqual(attributes, {"GET": 9}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/runtime/logs") + self.assertEqual(attributes, {"GET": 10}) + + +class TestBasicAuthUserConfigReader(TestHttpBase): + """ + Tests for user configuration reader. + """ + + @classmethod + def setUpClass(cls): + cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } + + def test_user_reader_valid_config(self): + file_content = ( + "# User configuration\n" + "user-1:70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145:\n" + "user-2:718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf:role1\n" + "user-3:280b73d2ac8375035501e5c06acb4b770e74ec95f479e6e2643227d7f57ce7ad:role1,role2\n" + ) + + with open(self.passwd_file, "w", encoding="utf-8") as passwd: + self.basic_auth_module._USERS.clear() + passwd.write(file_content) + + self.basic_auth_module._load_users() + + self.assertDictEqual( + self.basic_auth_module._USERS, + { + "user-1": [ + bytes.fromhex( + "70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145" + ), + 0b00, + ], + "user-2": [ + bytes.fromhex( + "718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf" + ), + 0b01, + ], + "user-3": [ + bytes.fromhex( + "280b73d2ac8375035501e5c06acb4b770e74ec95f479e6e2643227d7f57ce7ad" + ), + 0b11, + ], + }, + ) + + def test_user_reader_empty_user(self): + file_content = ( + ":70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145:\n" + ) + + with open(self.passwd_file, "w", encoding="utf-8") as passwd: + self.basic_auth_module._USERS.clear() + passwd.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_users() + + def test_user_reader_duplicate_user(self): + file_content = ( + "user-1:70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145:\n" + "user-1:718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf:\n" + ) + + with open(self.passwd_file, "w", encoding="utf-8") as passwd: + self.basic_auth_module._USERS.clear() + passwd.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_users() + + def test_user_reader_empty_password(self): + file_content = "user-1::\n" + + with open(self.passwd_file, "w", encoding="utf-8") as passwd: + self.basic_auth_module._USERS.clear() + passwd.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_users() + + +class TestBasicAuthRoleConfigReader(TestHttpBase): + """ + Tests for role configuration reader. + """ + + @classmethod + def setUpClass(cls): + cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } + + def test_role_reader_valid_config(self): + file_content = ( + "/*\n" + " *:*\n" + "# Comment\n" + "/app/resource\n" + "/app/resource/* # inline comment\n" + " GET: role_1\n" + " POST,PUT:role_2,role_3\n" + " OPTIONS: *# inline comment\n" + " DELETE:\n" + " \n" + " /app/api\n\n" + "GET: role_4 \n\n" + "POST , PUT : role_1 , role_2 " + ) + + with open(self.roles_file, "w", encoding="utf-8") as roles: + self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") + roles.write(file_content) + + self.basic_auth_module._load_roles() + + attributes = self.basic_auth_module._ATTR_TREE.get_attributes("/index.html") + self.assertEqual(attributes, {"*": self.basic_auth_module._NO_POLICY}) + + attributes = self.basic_auth_module._ATTR_TREE.get_attributes("/app/resource") + self.assertEqual( + attributes, + { + "GET": 0b001, + "POST": 0b110, + "PUT": 0b110, + "DELETE": 0b000, + "OPTIONS": self.basic_auth_module._NO_POLICY, + }, + ) + + attributes = self.basic_auth_module._ATTR_TREE.get_attributes("/app/api") + self.assertEqual( + attributes, + { + "GET": 0b1000, + "POST": 0b0011, + "PUT": 0b0011, + }, + ) + + def test_role_reader_missing_path(self): + file_content = " *:*\n" + + with open(self.roles_file, "w", encoding="utf-8") as roles: + self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") + roles.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_roles() + + def test_role_reader_duplicate_attribute(self): + file_content = "/app/resource\n" + " GET: role_1\n" + " GET: role_2\n" + + with open(self.roles_file, "w", encoding="utf-8") as roles: + self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") + roles.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_roles() + + +if __name__ == "__main__": + unittest.main(verbosity=2)