|
| 1 | +""" |
| 2 | +Admin utilities for the Streamlit template. |
| 3 | +
|
| 4 | +Provides functionality for admin-only operations like saving workspaces as demos. |
| 5 | +""" |
| 6 | + |
| 7 | +import hmac |
| 8 | +import shutil |
| 9 | +from pathlib import Path |
| 10 | + |
| 11 | +import streamlit as st |
| 12 | + |
| 13 | + |
| 14 | +def is_admin_configured() -> bool: |
| 15 | + """ |
| 16 | + Check if admin password is configured in Streamlit secrets. |
| 17 | +
|
| 18 | + Returns: |
| 19 | + bool: True if admin password is configured, False otherwise. |
| 20 | + """ |
| 21 | + try: |
| 22 | + return bool(st.secrets.get("admin", {}).get("password")) |
| 23 | + except (FileNotFoundError, KeyError): |
| 24 | + return False |
| 25 | + |
| 26 | + |
| 27 | +def verify_admin_password(password: str) -> bool: |
| 28 | + """ |
| 29 | + Verify the provided password against the configured admin password. |
| 30 | +
|
| 31 | + Uses constant-time comparison to prevent timing attacks. |
| 32 | +
|
| 33 | + Args: |
| 34 | + password: The password to verify. |
| 35 | +
|
| 36 | + Returns: |
| 37 | + bool: True if password matches, False otherwise. |
| 38 | + """ |
| 39 | + if not is_admin_configured(): |
| 40 | + return False |
| 41 | + |
| 42 | + try: |
| 43 | + stored_password = st.secrets["admin"]["password"] |
| 44 | + # Use constant-time comparison for security |
| 45 | + return hmac.compare_digest(password, stored_password) |
| 46 | + except (FileNotFoundError, KeyError): |
| 47 | + return False |
| 48 | + |
| 49 | + |
| 50 | +def get_demo_target_dir() -> Path: |
| 51 | + """ |
| 52 | + Get the directory where demo workspaces are stored. |
| 53 | +
|
| 54 | + Returns: |
| 55 | + Path: The demo workspaces directory. |
| 56 | + """ |
| 57 | + return Path("example-data/workspaces") |
| 58 | + |
| 59 | + |
| 60 | +def demo_exists(demo_name: str) -> bool: |
| 61 | + """ |
| 62 | + Check if a demo workspace with the given name already exists. |
| 63 | +
|
| 64 | + Args: |
| 65 | + demo_name: Name of the demo to check. |
| 66 | +
|
| 67 | + Returns: |
| 68 | + bool: True if demo exists, False otherwise. |
| 69 | + """ |
| 70 | + target_dir = get_demo_target_dir() |
| 71 | + demo_path = target_dir / demo_name |
| 72 | + return demo_path.exists() |
| 73 | + |
| 74 | + |
| 75 | +def _remove_directory_with_symlinks(path: Path) -> None: |
| 76 | + """ |
| 77 | + Remove a directory that may contain symlinks. |
| 78 | +
|
| 79 | + Handles symlinks properly by removing them without following. |
| 80 | +
|
| 81 | + Args: |
| 82 | + path: Path to the directory to remove. |
| 83 | + """ |
| 84 | + if not path.exists(): |
| 85 | + return |
| 86 | + |
| 87 | + for item in path.rglob("*"): |
| 88 | + if item.is_symlink(): |
| 89 | + item.unlink() |
| 90 | + |
| 91 | + # Now remove the rest normally |
| 92 | + if path.exists(): |
| 93 | + shutil.rmtree(path) |
| 94 | + |
| 95 | + |
| 96 | +def save_workspace_as_demo(workspace_path: Path, demo_name: str) -> tuple[bool, str]: |
| 97 | + """ |
| 98 | + Save the current workspace as a demo workspace. |
| 99 | +
|
| 100 | + Copies all files from the workspace to the demo directory, following symlinks |
| 101 | + to copy actual file contents rather than symlink references. |
| 102 | +
|
| 103 | + Args: |
| 104 | + workspace_path: Path to the source workspace. |
| 105 | + demo_name: Name for the new demo workspace. |
| 106 | +
|
| 107 | + Returns: |
| 108 | + tuple[bool, str]: (success, message) tuple indicating result. |
| 109 | + """ |
| 110 | + # Deferred import to avoid circular dependency with common.py |
| 111 | + from src.common.common import is_safe_workspace_name |
| 112 | + |
| 113 | + # Validate demo name |
| 114 | + if not demo_name: |
| 115 | + return False, "Demo name cannot be empty." |
| 116 | + |
| 117 | + if not is_safe_workspace_name(demo_name): |
| 118 | + return False, "Invalid demo name. Avoid path separators and special characters." |
| 119 | + |
| 120 | + # Validate source workspace exists |
| 121 | + if not workspace_path.exists(): |
| 122 | + return False, "Source workspace does not exist." |
| 123 | + |
| 124 | + # Get target directory |
| 125 | + target_dir = get_demo_target_dir() |
| 126 | + demo_path = target_dir / demo_name |
| 127 | + |
| 128 | + try: |
| 129 | + # Ensure parent directory exists |
| 130 | + target_dir.mkdir(parents=True, exist_ok=True) |
| 131 | + |
| 132 | + # Remove existing demo if it exists (handles symlinks properly) |
| 133 | + if demo_path.exists(): |
| 134 | + _remove_directory_with_symlinks(demo_path) |
| 135 | + |
| 136 | + # Copy workspace to demo directory, following symlinks to get actual files |
| 137 | + shutil.copytree( |
| 138 | + workspace_path, |
| 139 | + demo_path, |
| 140 | + symlinks=False, # Follow symlinks, copy actual files |
| 141 | + dirs_exist_ok=False |
| 142 | + ) |
| 143 | + |
| 144 | + return True, f"Workspace saved as demo '{demo_name}' successfully." |
| 145 | + |
| 146 | + except PermissionError: |
| 147 | + return False, "Permission denied. Cannot write to demo directory." |
| 148 | + except OSError as e: |
| 149 | + return False, f"Failed to save demo: {str(e)}" |
| 150 | + except Exception as e: |
| 151 | + return False, f"Unexpected error: {str(e)}" |
0 commit comments