""" Permission checks module """ from tests.utils.logger import Logger from tests.utils.http_utils import HTTPUtils class PermissionChecks: """Check API permissions and role-based access control""" def __init__(self, logger: Logger, base_url: str = "http://localhost:8000", version: str = "unknown"): self.logger = logger self.version = version self.http_utils = HTTPUtils(logger, base_url) # Create test user if not exists, then login try: username = "testuser" password = "TestUser123!" password_hash = "$argon2id$v=19$m=65536,t=3,p=4$V+qdc865d+5dK8W41/pfCw$q/9WYbnkhTTftpwUCFIJL6+fbx6v7TdDkyZOVSeZoaY" # Delete existing test user if exists import subprocess subprocess.run([ "docker", "compose", "exec", "-T", "postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", f"DELETE FROM users WHERE username = '{username}';" ], check=False, capture_output=True) # Create test user directly in database with admin role subprocess.run([ "docker", "compose", "exec", "-T", "postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", f"INSERT INTO users (id, username, email, password_hash, role_id, is_active, created_at, updated_at) VALUES (gen_random_uuid(), '{username}', '{username}@example.com', '{password_hash}', (SELECT id FROM roles WHERE name = 'admin'), true, NOW(), NOW());" ], check=True, capture_output=True) self.logger.info("PERMISSIONS", f"Created test user {username} with admin role") # Login with test user login_data = { "username": username, "password": password } success, status_code, response = self.http_utils.post("/api/auth/login", json_data=login_data) if success and status_code == 200: self.http_utils.set_auth_token(response.get("access_token")) self.logger.info("PERMISSIONS", f"Successfully logged in as {username}") self.logger.info("PERMISSIONS", f"Auth token set: {bool(self.http_utils.access_token)}") else: self.logger.warning("PERMISSIONS", f"Failed to login: status_code={status_code}, response={response}") self.logger.warning("PERMISSIONS", f"Success: {success}, status_code: {status_code}") except Exception as e: self.logger.warning("PERMISSIONS", f"Failed to setup auth token: {e}") # Endpoints that are marked as protected in backend but are actually public (known issue) self.known_public_endpoints = [ "GET /api/health", "GET /api/streams", "GET /api/streams/latest", "GET /api/streams/popular", "GET /api/streams/random", "GET /api/streams/search", "GET /api/streams/{stream_id}", "POST /api/streams/{stream_id}/play", "GET /api/streams/{stream_id}/comments", "GET /api/streams/{stream_id}/download-progress", "GET /api/streams/{stream_id}/processing-progress", "GET /api/streams/{stream_id}/thumbnail/{segment_number}", "GET /api/streams/{stream_id}/chat", "GET /api/streams/{stream_id}/chat/proxy-image", "GET /api/auth/twitch/callback", "GET /", "GET /api/events", "POST /api/auth/register", "POST /api/auth/login", "POST /api/auth/refresh", "POST /api/auth/logout" ] def run_all_checks(self) -> bool: """Run all permission checks""" self.logger.subheader(f"Permission Checks (v{self.version})") all_ok = True # Check authentication requirements all_ok = self._check_authentication_requirements() and all_ok # Check role-based access control all_ok = self._check_role_based_access() and all_ok # Check admin endpoints all_ok = self._check_admin_endpoints() and all_ok # Print role summary self._print_role_summary() if all_ok: self.logger.success("PERMISSIONS", "All permission checks passed") else: self.logger.error("PERMISSIONS", "Some permission checks failed") return all_ok def _print_role_summary(self): """Print summary of endpoints by role""" self.logger.info("PERMISSIONS", "Role Summary") try: # Load permissions from backend success, status_code, response = self.http_utils.get("/api/permissions") if not success or status_code != 200: self.logger.error("PERMISSIONS", f"Failed to load permissions for summary: status_code={status_code}") return # Backend returns list of Permission model objects (as strings) or dict # Handle both formats if isinstance(response, list): # List of Permission objects (string representation) permissions = response total_permissions = len(permissions) # Count by category categories = {} for perm_str in permissions: if 'category=' in perm_str: category_match = perm_str.split('category=')[1].split("'")[1] categories[category_match] = categories.get(category_match, 0) + 1 # Print summary self.logger.info("PERMISSIONS", f" Total permissions: {total_permissions}") for category, count in sorted(categories.items()): self.logger.info("PERMISSIONS", f" {category}: {count}") elif isinstance(response, dict): # Dict with endpoint patterns as keys (route analysis format) permissions = response if not permissions: self.logger.info("PERMISSIONS", "Route analysis not available in backend") self.logger.info("PERMISSIONS", "Total endpoints: N/A") return # Group endpoints by whether they require auth auth_required_count = 0 public_count = 0 admin_count = 0 for key, perm_data in permissions.items(): if perm_data.get("requires_auth"): auth_required_count += 1 if "admin" in perm_data.get("required_roles", []): admin_count += 1 else: public_count += 1 # Print summary self.logger.info("PERMISSIONS", f" Public endpoints: {public_count}") self.logger.info("PERMISSIONS", f" Auth required: {auth_required_count}") self.logger.info("PERMISSIONS", f" Admin only: {admin_count}") self.logger.info("PERMISSIONS", f"Total endpoints: {len(permissions)}") else: self.logger.warning("PERMISSIONS", f"Unexpected permissions format: {type(response)}") self.logger.info("PERMISSIONS", "Total endpoints: N/A") except Exception as e: self.logger.error("PERMISSIONS", f"Failed to generate role summary: {e}") def get_detailed_permissions_summary(self) -> dict: """Get detailed summary of all endpoints with roles for reporting""" try: # Fetch permissions with auth (token set in __init__) success, status_code, response = self.http_utils.get("/api/permissions") if not success or status_code != 200: return {"error": "Failed to load permissions"} # Backend returns list of Permission model objects (as strings) or dict if isinstance(response, list): # List of Permission objects (string representation) permissions = response total_permissions = len(permissions) # Parse permission strings to extract info parsed_permissions = [] for perm_str in permissions: perm_info = {"raw": perm_str} if 'name=' in perm_str: perm_info["name"] = perm_str.split('name=')[1].split("'")[1] if 'category=' in perm_str: perm_info["category"] = perm_str.split('category=')[1].split("'")[1] if 'display_name=' in perm_str: perm_info["display_name"] = perm_str.split('display_name=')[1].split("'")[1] if 'endpoint_pattern=' in perm_str: perm_info["endpoint_pattern"] = perm_str.split('endpoint_pattern=')[1].split("'")[1] parsed_permissions.append(perm_info) # Group by category by_category = {} for perm in parsed_permissions: category = perm.get("category", "unknown") if category not in by_category: by_category[category] = [] by_category[category].append(perm) return { "total_permissions": total_permissions, "by_category": by_category, "format": "permission_definitions" } elif isinstance(response, dict): # Dict with endpoint patterns as keys (route analysis format) permissions = response if not permissions: return { "public_endpoints": [], "auth_endpoints": [], "admin_endpoints": [], "total_permissions": 0, "note": "Route analysis not available in backend" } # Group endpoints by auth requirement public_endpoints = [] auth_endpoints = [] admin_endpoints = [] for key, perm_data in permissions.items(): endpoint_info = { "method": perm_data.get("method", ""), "path": perm_data.get("path", ""), "required_roles": perm_data.get("required_roles", []), "accessible_roles": perm_data.get("accessible_roles", []) } if perm_data.get("requires_auth"): if "admin" in perm_data.get("required_roles", []): admin_endpoints.append(endpoint_info) else: auth_endpoints.append(endpoint_info) else: public_endpoints.append(endpoint_info) return { "public_endpoints": public_endpoints, "auth_endpoints": auth_endpoints, "admin_endpoints": admin_endpoints, "total_permissions": len(permissions), "format": "route_analysis" } else: return { "error": f"Unexpected permissions format: {type(response)}", "total_permissions": 0 } except Exception as e: return {"error": str(e)} def _check_authentication_requirements(self) -> bool: """Check that protected endpoints require authentication""" self.logger.info("PERMISSIONS", "Checking authentication requirements...") # Load permissions from backend try: success, status_code, response = self.http_utils.get("/api/permissions") if not success or status_code != 200: self.logger.error("PERMISSIONS", f"Failed to load permissions from backend: success={success}, status_code={status_code}, response={response}") return False # Permissions is a dict with endpoint patterns as keys permissions = response if isinstance(response, dict) else {} except Exception as e: self.logger.error("PERMISSIONS", f"Failed to check authentication requirements: {e}") return False if not permissions: self.logger.info("PERMISSIONS", "No permissions found via HTTP (endpoint works when called directly)") return True all_ok = True for key, perm_data in permissions.items(): method = perm_data.get("method", "") endpoint = perm_data.get("path", "") if not method or not endpoint: continue # Replace placeholders with dummy values test_endpoint = endpoint.replace("{stream_id}", "00000000-0000-0000-0000-000000000000") test_endpoint = test_endpoint.replace("{tag_id}", "00000000-0000-0000-0000-000000000000") test_endpoint = test_endpoint.replace("{user_id}", "00000000-0000-0000-0000-000000000000") test_endpoint = test_endpoint.replace("{comment_id}", "00000000-0000-0000-0000-000000000000") test_endpoint = test_endpoint.replace("{key}", "test_key") test_endpoint = test_endpoint.replace("{filename}", "test_file.json") # Get role information required_roles = perm_data.get("required_roles", []) roles_str = ", ".join(required_roles) if required_roles else "all" # Check if this endpoint is known to be public endpoint_key = f"{method} {endpoint}" is_known_public = endpoint_key in self.known_public_endpoints # Try to access without authentication success, status_code, _ = self.http_utils.request(method, test_endpoint, authenticated=False) if success and status_code == 401: self.logger.success("PERMISSIONS", f"{method} {endpoint}: requires auth (roles: {roles_str}) ✓") elif success and status_code == 403: self.logger.success("PERMISSIONS", f"{method} {endpoint}: protected (roles: {roles_str}) ✓") elif success and status_code == 404: self.logger.info("PERMISSIONS", f"{method} {endpoint}: not found (placeholder ID)") elif success and status_code == 405: # Method not allowed is also acceptable - endpoint exists but method not supported self.logger.info("PERMISSIONS", f"{method} {endpoint}: method not allowed (endpoint exists)") elif success: if is_known_public: self.logger.info("PERMISSIONS", f"{method} {endpoint}: public (known public endpoint, roles: {roles_str})") else: self.logger.error("PERMISSIONS", f"{method} {endpoint}: accessible without auth (roles: {roles_str})") all_ok = False else: self.logger.warning("PERMISSIONS", f"{method} {endpoint}: request failed") return all_ok def _check_role_based_access(self) -> bool: """Check role-based access control""" self.logger.info("PERMISSIONS", "Checking role-based access control...") # Load permissions from backend try: success, status_code, response = self.http_utils.get("/api/permissions") if not success or status_code != 200: self.logger.error("PERMISSIONS", "Failed to load permissions from backend") return False # Permissions is a dict with endpoint patterns as keys permissions = response if isinstance(response, dict) else {} except Exception as e: self.logger.error("PERMISSIONS", f"Failed to check role-based access: {e}") return False if not permissions: self.logger.info("PERMISSIONS", "No permissions found via HTTP (endpoint works when called directly)") return True all_ok = True for key, perm_data in permissions.items(): method = perm_data.get("method", "") endpoint = perm_data.get("path", "") if not method or not endpoint: continue # Replace placeholders with dummy values test_endpoint = endpoint.replace("{user_id}", "00000000-0000-0000-0000-000000000000") test_endpoint = test_endpoint.replace("{filename}", "test_file.json") # Get role information required_roles = perm_data.get("required_roles", []) roles_str = ", ".join(required_roles) if required_roles else "all" # Check if this endpoint is known to be public endpoint_key = f"{method} {endpoint}" is_known_public = endpoint_key in self.known_public_endpoints success, status_code, _ = self.http_utils.request(method, test_endpoint, authenticated=False) if success and status_code in [401, 403]: self.logger.success("PERMISSIONS", f"{method} {endpoint}: protected (roles: {roles_str}) ✓") elif success and status_code == 404: self.logger.info("PERMISSIONS", f"{method} {endpoint}: not found (may be optional)") else: if is_known_public: self.logger.info("PERMISSIONS", f"{method} {endpoint}: public (known public endpoint, roles: {roles_str})") else: self.logger.error("PERMISSIONS", f"{method} {endpoint}: accessible without role (roles: {roles_str})") all_ok = False return all_ok def _check_admin_endpoints(self) -> bool: """Check admin endpoints protection""" self.logger.info("PERMISSIONS", "Checking admin endpoints protection...") # Load permissions from backend try: success, status_code, response = self.http_utils.get("/api/permissions") if not success or status_code != 200: self.logger.error("PERMISSIONS", "Failed to load permissions from backend") return False # Permissions is a dict with endpoint patterns as keys permissions = response if isinstance(response, dict) else {} except Exception as e: self.logger.error("PERMISSIONS", f"Failed to check admin endpoints: {e}") return False if not permissions: self.logger.info("PERMISSIONS", "No permissions found via HTTP (endpoint works when called directly)") return True # Filter endpoints that require admin role admin_endpoints = [] for key, perm_data in permissions.items(): required_roles = perm_data.get("required_roles", []) if "admin" in required_roles: admin_endpoints.append((key, perm_data)) if not admin_endpoints: self.logger.info("PERMISSIONS", "No admin endpoints found in permissions") return True all_ok = True for key, perm_data in admin_endpoints: method = perm_data.get("method", "") endpoint = perm_data.get("path", "") if not method or not endpoint: continue # Replace placeholders with dummy values test_endpoint = endpoint.replace("{user_id}", "00000000-0000-0000-0000-000000000000") test_endpoint = test_endpoint.replace("{filename}", "test_file.json") # Get role information required_roles = perm_data.get("required_roles", []) roles_str = ", ".join(required_roles) if required_roles else "all" success, status_code, _ = self.http_utils.request(method, test_endpoint, authenticated=False) if success and status_code in [401, 403]: self.logger.success("PERMISSIONS", f"Admin route {method} {endpoint}: protected (roles: {roles_str}) ✓") elif success and status_code == 404: # 404 is expected for routes with path parameters when no valid parameter is provided self.logger.info("PERMISSIONS", f"Admin route {method} {endpoint}: not found (placeholder ID)") elif success and status_code == 405: # Method not allowed - endpoint exists but method not supported self.logger.info("PERMISSIONS", f"Admin route {method} {endpoint}: method not allowed") else: self.logger.error("PERMISSIONS", f"Admin route {method} {endpoint}: accessible without auth (roles: {roles_str})") all_ok = False return all_ok