mirror of
https://github.com/cisagov/ScubaGoggles.git
synced 2026-07-17 00:47:28 +02:00
Creates config UI (#945)
Run Link Checker for Cron Job / markdown-link-check (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Pylint (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Run OPA Tests (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Run Markdown Link Check (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Run Python Tests (push) Has been cancelled
Run Smoke Test / configuration (push) Has been cancelled
Run Smoke Test / smoke-test (push) Has been cancelled
Run Smoke Test / lhci (push) Has been cancelled
Update OPA Version Dependency / Check for OPA Update (push) Has been cancelled
Update OPA Version Dependency / Update OPA and Create PR (push) Has been cancelled
Run Link Checker for Cron Job / markdown-link-check (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Pylint (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Run OPA Tests (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Run Markdown Link Check (push) Has been cancelled
Run CI Pipeline for ScubaGoggles / Run Python Tests (push) Has been cancelled
Run Smoke Test / configuration (push) Has been cancelled
Run Smoke Test / smoke-test (push) Has been cancelled
Run Smoke Test / lhci (push) Has been cancelled
Update OPA Version Dependency / Check for OPA Update (push) Has been cancelled
Update OPA Version Dependency / Update OPA and Create PR (push) Has been cancelled
* Updates to scubagoggles config UI * Updates to placeholders * updating placeholder for organization unit * Fixing linter * fixing linter * Updating more linter * updating linter * Linter fixes pt 5 * linter fix pt 6 * linter fixes pt 7 * cleaning up unnecessary code * fixing linter pt 7 * fixing linter pt 10 * Updating typo * Removed advanced tab references throughout documents * update validation * removed unnecessary code * Updating readme * Adding pylint ignores * fixed linter * added dns config * Updated based on PR feedback * updated 2 more comments * removed unnecessary import * Switched Annotate and Omit locations * Updated save button * Added static method for loading backend * added name normailzation for configu ui * Updated tabs and added IMAP exception * fixing linter * Added shorthand support for config * Added warning message for no products and fixed alignment * updated regex * Added dark mode flag for config ui * Fixed dark mode * added pylint ignore * Updates based on PR comments * Adds pylint * Added advanced tab for additional config options * fixed linter * Adding Readme mention * Removed pywebview, and refactored code * fixing linter * Added preferreddohservers config * add linter ignore * Fixed bugs * Updated Advanced help dialog * Removed depreciating package
This commit is contained in:
@@ -46,6 +46,7 @@ We use a three-step process:
|
||||
|
||||
- [Usage: Parameters](https://github.com/cisagov/ScubaGoggles/blob/main/docs/usage/Parameters.md)
|
||||
- [Usage: Config File](https://github.com/cisagov/ScubaGoggles/blob/main/docs/usage/Config.md)
|
||||
- [Configuration UI](https://github.com/cisagov/ScubaGoggles/blob/main/scubagoggles/ui/README.md) — web-based form for building config files
|
||||
- [Usage: Examples](https://github.com/cisagov/ScubaGoggles/blob/main/docs/usage/Examples.md)
|
||||
- [Reviewing Output](https://github.com/cisagov/ScubaGoggles/blob/main/docs/usage/ReviewOutput.md)
|
||||
- [Limitations](https://github.com/cisagov/ScubaGoggles/blob/main/docs/usage/Limitations.md)
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
|
||||
# Usage: Config File
|
||||
|
||||
> [!TIP]
|
||||
> Prefer a graphical interface? The [Configuration UI](../../scubagoggles/ui/README.md) provides a web-based form for building config files with validation, policy browsing, and YAML export.
|
||||
|
||||
All ScubaGoggles [parameters](Parameters.md) can be placed into a configuration file in order to made execution easier. The path of the file is specified by the `--config` parameter, and its contents are expected as YAML.
|
||||
|
||||
> [!NOTE]
|
||||
|
||||
@@ -26,4 +26,8 @@ MarkupSafe>=2.1.5
|
||||
pyyaml>=6.0.2
|
||||
requests>=2.32.3
|
||||
tqdm>=4.66.5
|
||||
|
||||
packaging>=26.0
|
||||
|
||||
# UI dependencies for the Streamlit-based configuration interface
|
||||
streamlit>=1.35.0
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
# ScubaGoggles Configuration UI
|
||||
|
||||
This directory contains a professional Streamlit-based configuration interface for ScubaGoggles, inspired by ScubaGear's ScubaConfigApp but built with Python and web technologies.
|
||||
|
||||
## Overview
|
||||
|
||||
The ScubaGoggles Configuration UI provides a comprehensive web interface for creating ScubaGoggles configuration files to manage policy exclusions, annotations, and baseline control settings. The interface features:
|
||||
|
||||
- **Organization Information Management**: Configure organization details and descriptions
|
||||
- **Product/Baseline Selection**: Visual selection of Google Workspace products to assess (11 products available)
|
||||
- **Policy Omission**: Exclude specific policies with documented rationale and expiration dates
|
||||
- **Policy Annotation**: Add comments, mark incorrect results, and set remediation dates
|
||||
- **Exclusions**: Define break glass (emergency access) accounts and IMAP exceptions
|
||||
- **Configuration Preview & Export**: View and download YAML configuration files
|
||||
- **Dark Mode Support**: Automatically follows browser/OS dark mode preference, or use `--dark` / `SCUBAGOGGLES_UI_DARK` to force dark mode
|
||||
|
||||
## Features
|
||||
|
||||
The application consolidates all configuration capabilities into a single, professional interface with the following features:
|
||||
|
||||
- **Organization Configuration**: Configure organization name (required), unit name, and assessment description
|
||||
- **Product Selection**: Choose from 11 Google Workspace baselines (Common Controls, Assured Controls, Gmail, Drive, Calendar, Meet, Groups, Chat, Sites, Classroom, Gemini) with icons, policy counts, and Select All/Clear All functionality
|
||||
- **Policy Omission Management**: Exclude specific policies from assessment with documented rationale, optional expiration dates, and summary views
|
||||
- **Policy Annotation**: Add comments and documentation to policies, mark incorrect results, set remediation dates with visual status indicators
|
||||
- **Break Glass Account Configuration**: Define and manage super admin emergency access accounts with email validation
|
||||
- **IMAP Exception Configuration**: Define OUs and groups where IMAP access is allowed per GWS.GMAIL.9.1
|
||||
- **Authentication Settings**: Configure Service Account credentials (Customer ID, Subject Email, JSON file path), OAuth 2.0, or Application Default Credentials
|
||||
- **Output & Execution Options**: Set output directory, report formats, quiet mode, and dark mode preferences
|
||||
- **Configuration Management**: Import existing YAML files, preview generated configuration with validation, and download formatted YAML files
|
||||
- **Visual Feedback**: Real-time status indicators (green dots for configured items, orange for editing), selection summaries, and comprehensive validation
|
||||
|
||||
## Files
|
||||
|
||||
- `scubaconfigapp.py` - Complete professional configuration application with all features
|
||||
- `launch.py` - Launcher script that starts the Streamlit server and opens the browser
|
||||
- `validation.py` - Configuration validation utilities
|
||||
- `__init__.py` - Package initialization
|
||||
|
||||
## Installation
|
||||
|
||||
### Prerequisites
|
||||
|
||||
1. **ScubaGoggles** must be installed and available
|
||||
2. **Python 3.10+** is required
|
||||
3. **Streamlit** UI dependency
|
||||
|
||||
### Install Dependencies
|
||||
|
||||
```bash
|
||||
# Install all requirements including UI dependencies (from the ScubaGoggles directory)
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
### Method 1: Direct Launch (Recommended)
|
||||
|
||||
```bash
|
||||
# From the ScubaGoggles root directory
|
||||
python -m scubagoggles.ui.launch
|
||||
```
|
||||
|
||||
This will start the Streamlit server and open the app in your default web browser. Dark mode is automatically detected from your browser's preferred color scheme via CSS media queries.
|
||||
|
||||
### Forcing dark mode
|
||||
|
||||
To force dark mode regardless of browser settings, use the `--dark` flag:
|
||||
|
||||
```bash
|
||||
python -m scubagoggles.ui.launch --dark
|
||||
```
|
||||
|
||||
Alternatively, set the `SCUBAGOGGLES_UI_DARK` environment variable. Truthy values: `1`, `true`, `yes`, `on` (case-insensitive):
|
||||
|
||||
```bash
|
||||
# Linux / macOS
|
||||
export SCUBAGOGGLES_UI_DARK=1
|
||||
python -m scubagoggles.ui.launch
|
||||
|
||||
# Windows (PowerShell)
|
||||
$env:SCUBAGOGGLES_UI_DARK = "1"
|
||||
python -m scubagoggles.ui.launch
|
||||
```
|
||||
|
||||
If you use **Method 2** (Streamlit only) and want to force dark mode, set the environment variable and pass the Streamlit theme flag:
|
||||
|
||||
```bash
|
||||
export SCUBAGOGGLES_UI_DARK=1
|
||||
streamlit run scubagoggles/ui/scubaconfigapp.py --theme.base dark
|
||||
```
|
||||
|
||||
### Method 2: Streamlit Command
|
||||
|
||||
```bash
|
||||
# From the ScubaGoggles root directory
|
||||
streamlit run scubagoggles/ui/scubaconfigapp.py
|
||||
```
|
||||
|
||||
## Interface Overview
|
||||
|
||||
### Navigation Tabs
|
||||
|
||||
1. **Main** - Configure organization information and select products to assess
|
||||
2. **Annotate Policies** - Add comments, mark incorrect results, and set remediation dates
|
||||
3. **Omit Policies** - Exclude specific policies with documented rationale
|
||||
4. **Exclusions** - Configure break glass accounts and IMAP exceptions
|
||||
5. **DNS Configuration** - Configure DNS resolvers and DoH fallback settings
|
||||
6. **Advanced** - Configure strict mode, report options, and fail-fast behavior
|
||||
7. **Preview** - Review and download the generated YAML configuration
|
||||
|
||||
### Header Toolbar
|
||||
|
||||
- **Open**: Import existing YAML configuration files via the browser file picker
|
||||
- **Reset**: Reset all fields to defaults (with confirmation dialog)
|
||||
- **Help**: Access comprehensive help and documentation
|
||||
|
||||
### Key Interface Elements
|
||||
|
||||
- **Status Indicators**: Green dots indicate configured items, orange dots show items being edited
|
||||
- **Context Help**: Expandable help sections in each tab with guidelines and best practices
|
||||
- **Validation**: Real-time validation ensures required fields are completed before export
|
||||
|
||||
## Configuration Workflow
|
||||
|
||||
1. **Start the UI**: Launch using one of the methods above
|
||||
2. **Organization Information**: Enter organization name (required) and unit name in the Main tab
|
||||
3. **Select Products**: Choose which Google Workspace products to assess (at least one required)
|
||||
4. **Annotate Policies** (Optional): Add comments and documentation to policy results
|
||||
5. **Omit Policies** (Optional): Exclude specific policies with documented rationale
|
||||
6. **Exclusions** (Optional): Configure break glass accounts and IMAP exceptions
|
||||
7. **DNS Configuration** (Optional): Configure preferred DNS resolvers and DoH fallback behavior
|
||||
8. **Advanced** (Optional): Configure strict mode, report output, and fail-fast behavior
|
||||
9. **Preview & Export**: Review the generated configuration and download the YAML file
|
||||
|
||||
### Typical Use Cases
|
||||
|
||||
- **Creating a New Configuration**: Start with Main tab, select products, then Preview & Export
|
||||
- **Managing Policy Exceptions**: Select products, go to Omit Policies tab, configure exclusions with rationale
|
||||
- **Documentation & Remediation**: Select products, go to Annotate Policies tab, add comments and set remediation dates
|
||||
- **Importing Existing Configuration**: Use Open button in header to import, review/modify settings, then export updated configuration
|
||||
|
||||
## Development
|
||||
|
||||
### Adding New Features
|
||||
|
||||
1. **UI Components**: Add to respective tab rendering methods
|
||||
2. **Validation**: Extend `validation.py` with new validators
|
||||
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
1. **ScubaGoggles Not Found**
|
||||
- Ensure ScubaGoggles is installed: `pip install scubagoggles`
|
||||
- Check Python path includes ScubaGoggles installation directory
|
||||
- Run from the ScubaGoggles root directory
|
||||
|
||||
2. **Streamlit Not Available**
|
||||
- Install requirements: `pip install -r requirements.txt`
|
||||
- Or install directly: `pip install streamlit`
|
||||
- Verify installation: `streamlit --version`
|
||||
|
||||
3. **Port Already in Use**
|
||||
- Streamlit default port (8501) may be in use
|
||||
- Specify different port: `streamlit run scubagoggles/ui/scubaconfigapp.py --server.port 8502`
|
||||
|
||||
4. **Module Import Errors**
|
||||
- Ensure you're running from the ScubaGoggles root directory
|
||||
- Check that `scubagoggles` package is in your Python path
|
||||
- Verify ScubaGoggles baseline files exist in `scubagoggles/baselines/`
|
||||
|
||||
5. **File Upload/Configuration Import Issues**
|
||||
- Check file permissions for uploaded files
|
||||
- Ensure YAML files are properly formatted
|
||||
- Verify configuration file schema matches ScubaGoggles requirements
|
||||
|
||||
### Getting Help
|
||||
|
||||
- [ScubaGoggles Documentation](https://github.com/cisagov/ScubaGoggles)
|
||||
- [Streamlit Documentation](https://docs.streamlit.io/)
|
||||
- [Report Issues](https://github.com/cisagov/ScubaGoggles/issues)
|
||||
|
||||
## License
|
||||
|
||||
This code is part of ScubaGoggles and follows the same license terms.
|
||||
|
||||
## Contributing
|
||||
|
||||
Contributions are welcome! Please see the main ScubaGoggles repository for contribution guidelines.
|
||||
@@ -0,0 +1,15 @@
|
||||
"""
|
||||
ScubaGoggles UI Package
|
||||
Configuration interface for ScubaGoggles using Streamlit
|
||||
"""
|
||||
|
||||
__version__ = "1.0.0"
|
||||
__author__ = "CISA SCuBA Team"
|
||||
|
||||
from .scubaconfigapp import ScubaConfigApp
|
||||
from .validation import ConfigValidator
|
||||
|
||||
__all__ = [
|
||||
"ScubaConfigApp",
|
||||
"ConfigValidator",
|
||||
]
|
||||
@@ -0,0 +1,135 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
ScubaGoggles UI Launcher
|
||||
|
||||
Starts the Streamlit-based configuration app in the user's default browser.
|
||||
Dark mode is auto-detected by the browser via CSS media queries. Use --dark
|
||||
or SCUBAGOGGLES_UI_DARK=1 to force dark mode for both Streamlit widgets and
|
||||
the app's custom CSS.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import atexit
|
||||
import importlib.util
|
||||
import os
|
||||
import signal
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _find_free_port() -> int:
|
||||
"""Find an available port on localhost."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.bind(("localhost", 0))
|
||||
return s.getsockname()[1]
|
||||
|
||||
|
||||
def _resolve_app_file() -> Path | None:
|
||||
"""Return the path to the Streamlit app file."""
|
||||
candidate = Path(__file__).parent / "scubaconfigapp.py"
|
||||
return candidate if candidate.exists() else None
|
||||
|
||||
|
||||
def _is_streamlit_installed() -> bool:
|
||||
"""Return True if the streamlit package is importable."""
|
||||
return importlib.util.find_spec("streamlit") is not None
|
||||
|
||||
|
||||
def _kill_process_tree(pid: int) -> None:
|
||||
"""Kill a process and all its children by PID."""
|
||||
try:
|
||||
if sys.platform == "win32":
|
||||
subprocess.run(
|
||||
["taskkill", "/F", "/T", "/PID", str(pid)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
check=False,
|
||||
)
|
||||
else:
|
||||
os.killpg(os.getpgid(pid), signal.SIGKILL)
|
||||
except (ProcessLookupError, OSError):
|
||||
pass
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Launch the ScubaGoggles UI in the default web browser."""
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description="ScubaGoggles Configuration UI (Streamlit)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dark",
|
||||
action="store_true",
|
||||
help="Force dark theme (overrides browser preference)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
if args.dark:
|
||||
os.environ["SCUBAGOGGLES_UI_DARK"] = "1"
|
||||
|
||||
force_dark = os.environ.get(
|
||||
"SCUBAGOGGLES_UI_DARK", "",
|
||||
).strip().lower() in ("1", "true", "yes", "on")
|
||||
|
||||
app_to_run = _resolve_app_file()
|
||||
if not app_to_run:
|
||||
print("No UI application found!")
|
||||
print("Please ensure the UI modules are properly installed.")
|
||||
sys.exit(1)
|
||||
|
||||
if not _is_streamlit_installed():
|
||||
print("Streamlit is not installed!")
|
||||
print("Please install requirements:")
|
||||
print(" pip install -r requirements.txt")
|
||||
sys.exit(1)
|
||||
|
||||
port = _find_free_port()
|
||||
|
||||
cmd = [
|
||||
sys.executable, "-m", "streamlit", "run",
|
||||
str(app_to_run),
|
||||
"--server.address", "localhost",
|
||||
"--server.port", str(port),
|
||||
"--server.headless", "false",
|
||||
"--browser.gatherUsageStats", "false",
|
||||
]
|
||||
if force_dark:
|
||||
cmd += ["--theme.base", "dark"]
|
||||
|
||||
popen_kwargs: dict = {}
|
||||
if sys.platform != "win32":
|
||||
# On Unix, start a new session so os.killpg can target the group.
|
||||
# On Windows we intentionally stay in the same console group so
|
||||
# that Ctrl+C propagates naturally to both parent and child.
|
||||
popen_kwargs["start_new_session"] = True
|
||||
|
||||
theme_label = "dark (forced)" if force_dark else "auto"
|
||||
print(
|
||||
"Starting ScubaGoggles Configuration UI "
|
||||
f"({theme_label} theme) ...",
|
||||
)
|
||||
print(f"Opening http://localhost:{port} in your browser.")
|
||||
print("Press Ctrl+C to stop the server.\n")
|
||||
|
||||
with subprocess.Popen(cmd, **popen_kwargs) as server_process:
|
||||
# Register cleanup so the Streamlit process tree is always killed —
|
||||
# even if the terminal window is closed or the parent exits unexpectedly.
|
||||
atexit.register(_kill_process_tree, server_process.pid)
|
||||
|
||||
if sys.platform != "win32":
|
||||
signal.signal(
|
||||
signal.SIGINT,
|
||||
lambda *_: (_kill_process_tree(server_process.pid), sys.exit(0)),
|
||||
)
|
||||
|
||||
try:
|
||||
server_process.wait()
|
||||
except KeyboardInterrupt:
|
||||
print("\nScubaGoggles UI stopped by user")
|
||||
finally:
|
||||
_kill_process_tree(server_process.pid)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,258 @@
|
||||
"""
|
||||
Configuration validation utilities for ScubaGoggles UI.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
class ConfigValidator:
|
||||
"""Handles validation of ScubaGoggles configuration parameters"""
|
||||
|
||||
@staticmethod
|
||||
def validate_credentials_file(file_path: str) -> Tuple[bool, Optional[str]]:
|
||||
"""Validate Google service account credentials file"""
|
||||
error: Optional[str] = None
|
||||
|
||||
if not file_path:
|
||||
error = "Credentials file path is required"
|
||||
|
||||
path = Path(file_path) if not error else None
|
||||
if path and not path.exists():
|
||||
error = f"Credentials file does not exist: {file_path}"
|
||||
|
||||
if path and path.suffix.lower() != ".json":
|
||||
error = "Credentials file must be a JSON file"
|
||||
|
||||
try:
|
||||
if not error and path is not None:
|
||||
with open(path, "r", encoding="utf-8") as file:
|
||||
creds_data = json.load(file)
|
||||
else:
|
||||
creds_data = None
|
||||
except json.JSONDecodeError as exc:
|
||||
error = f"Invalid JSON in credentials file: {exc}"
|
||||
creds_data = None
|
||||
except OSError as exc:
|
||||
error = f"Error reading credentials file: {exc}"
|
||||
creds_data = None
|
||||
|
||||
if not error and creds_data is not None:
|
||||
required_fields = ["type", "client_id", "client_email", "private_key"]
|
||||
missing_fields = [
|
||||
field for field in required_fields if field not in creds_data
|
||||
]
|
||||
|
||||
if missing_fields:
|
||||
error = (
|
||||
"Credentials file missing required fields: "
|
||||
f"{', '.join(missing_fields)}"
|
||||
)
|
||||
elif creds_data.get("type") != "service_account":
|
||||
error = "Credentials file must be for a service account"
|
||||
|
||||
return error is None, error
|
||||
|
||||
@staticmethod
|
||||
def validate_access_token(token: str) -> Tuple[bool, Optional[str]]:
|
||||
"""Validate access token format"""
|
||||
if not token:
|
||||
return False, "Access token is required"
|
||||
|
||||
# Basic token format validation.
|
||||
# Google tokens are typically long alphanumeric strings.
|
||||
if len(token) < 50:
|
||||
return False, "Access token appears to be too short"
|
||||
|
||||
# Check for common token patterns
|
||||
if not re.match(r"^[A-Za-z0-9._-]+$", token):
|
||||
return False, "Access token contains invalid characters"
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def validate_output_path(output_path: str) -> Tuple[bool, Optional[str]]:
|
||||
"""Validate output directory path"""
|
||||
if not output_path:
|
||||
return False, "Output path is required"
|
||||
|
||||
path = Path(output_path)
|
||||
|
||||
# Check if parent directory exists
|
||||
if not path.parent.exists():
|
||||
return False, f"Parent directory does not exist: {path.parent}"
|
||||
|
||||
# Check if path is writable
|
||||
try:
|
||||
# Try to create the directory if it doesn't exist
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Test write permissions
|
||||
test_file = path / "test_write.tmp"
|
||||
test_file.touch()
|
||||
test_file.unlink()
|
||||
|
||||
return True, None
|
||||
except PermissionError:
|
||||
return False, f"No write permission for output path: {output_path}"
|
||||
except Exception as exc:
|
||||
return False, f"Error with output path: {exc}"
|
||||
|
||||
@staticmethod
|
||||
def validate_baselines(
|
||||
baselines: List[str],
|
||||
available_baselines: List[str],
|
||||
) -> Tuple[bool, Optional[str]]:
|
||||
"""Validate selected baselines"""
|
||||
if not baselines:
|
||||
return False, "At least one baseline must be selected"
|
||||
|
||||
invalid_baselines = [baseline for baseline in baselines
|
||||
if baseline not in available_baselines]
|
||||
if invalid_baselines:
|
||||
return False, (
|
||||
"Invalid baselines selected: "
|
||||
f"{', '.join(invalid_baselines)}"
|
||||
)
|
||||
|
||||
return True, None
|
||||
|
||||
EMAIL_PATTERN = re.compile(
|
||||
r"^[a-zA-Z0-9_%+-]+(\.[a-zA-Z0-9_%+-]+)*"
|
||||
r"@"
|
||||
r"[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?"
|
||||
r"(\.[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?)*"
|
||||
r"\.[a-zA-Z]{2,}$"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def validate_break_glass_accounts(
|
||||
accounts: List[str],
|
||||
) -> Tuple[bool, Optional[str]]:
|
||||
"""Validate break glass account email addresses"""
|
||||
if not accounts:
|
||||
return True, None # Optional field
|
||||
|
||||
invalid_emails: List[str] = []
|
||||
for account in accounts:
|
||||
if not ConfigValidator.EMAIL_PATTERN.match(account.strip()):
|
||||
invalid_emails.append(account)
|
||||
|
||||
if invalid_emails:
|
||||
return False, (
|
||||
"Invalid email format for break glass accounts: "
|
||||
f"{', '.join(invalid_emails)}"
|
||||
)
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def validate_email(email: str) -> bool:
|
||||
"""Return True if email has a valid format."""
|
||||
return bool(ConfigValidator.EMAIL_PATTERN.match(email.strip()))
|
||||
|
||||
@staticmethod
|
||||
def validate_imap_exceptions(
|
||||
exceptions: List[Dict[str, Any]],
|
||||
) -> Tuple[bool, Optional[str]]:
|
||||
"""Validate IMAP exception entries."""
|
||||
if not exceptions:
|
||||
return True, None
|
||||
|
||||
errors: List[str] = []
|
||||
for i, entry in enumerate(exceptions):
|
||||
if not isinstance(entry, dict):
|
||||
errors.append(f"Entry {i + 1}: must be a mapping")
|
||||
continue
|
||||
ou = entry.get('ou', '').strip() if entry.get('ou') else ''
|
||||
group = entry.get('group', '').strip() if entry.get('group') else ''
|
||||
if not ou and not group:
|
||||
errors.append(
|
||||
f"Entry {i + 1}: at least an OU or group is required"
|
||||
)
|
||||
if group and not ConfigValidator.EMAIL_PATTERN.match(group):
|
||||
errors.append(
|
||||
f"Entry {i + 1}: invalid group email format: {group}"
|
||||
)
|
||||
|
||||
if errors:
|
||||
return False, "; ".join(errors)
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def validate_tenant_domain(domain: str) -> Tuple[bool, Optional[str]]:
|
||||
"""Validate tenant domain format"""
|
||||
if not domain:
|
||||
return True, None # Optional field
|
||||
|
||||
# Basic domain validation
|
||||
domain_pattern = re.compile(
|
||||
r"^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?"
|
||||
r"(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*$",
|
||||
)
|
||||
|
||||
if not domain_pattern.match(domain):
|
||||
return False, "Invalid domain format"
|
||||
|
||||
return True, None
|
||||
|
||||
@staticmethod
|
||||
def _validate_auth(
|
||||
config_dict: Dict[str, Any],
|
||||
errors: List[str],
|
||||
):
|
||||
"""Validate authentication fields and append any errors."""
|
||||
has_creds = config_dict.get("credentials")
|
||||
has_token = config_dict.get("accesstoken")
|
||||
|
||||
if not has_creds and not has_token:
|
||||
errors.append("Either credentials file or access token is required")
|
||||
elif has_creds:
|
||||
is_valid, error = ConfigValidator.validate_credentials_file(has_creds)
|
||||
if not is_valid:
|
||||
errors.append(f"Credentials validation: {error}")
|
||||
elif has_token:
|
||||
is_valid, error = ConfigValidator.validate_access_token(has_token)
|
||||
if not is_valid:
|
||||
errors.append(f"Access token validation: {error}")
|
||||
|
||||
@staticmethod
|
||||
def validate_complete_config(
|
||||
config_dict: Dict[str, Any],
|
||||
available_baselines: List[str],
|
||||
) -> Tuple[bool, List[str]]:
|
||||
"""Validate complete configuration and return all errors"""
|
||||
errors: List[str] = []
|
||||
|
||||
ConfigValidator._validate_auth(config_dict, errors)
|
||||
|
||||
# Baseline validation
|
||||
baselines = config_dict.get("baselines", [])
|
||||
is_valid, error = ConfigValidator.validate_baselines(
|
||||
baselines,
|
||||
available_baselines,
|
||||
)
|
||||
if not is_valid:
|
||||
errors.append(f"Baseline validation: {error}")
|
||||
|
||||
# Optional field validators: (config key, default, validator, label)
|
||||
_optional_checks: List[tuple] = [
|
||||
("outputpath", None, ConfigValidator.validate_output_path,
|
||||
"Output path validation"),
|
||||
("breakglassaccounts", [], ConfigValidator.validate_break_glass_accounts,
|
||||
"Break glass accounts validation"),
|
||||
("imapexceptions", [], ConfigValidator.validate_imap_exceptions,
|
||||
"IMAP exceptions validation"),
|
||||
("tenant", None, ConfigValidator.validate_tenant_domain,
|
||||
"Tenant domain validation"),
|
||||
]
|
||||
for key, default, validator, label in _optional_checks:
|
||||
value = config_dict.get(key, default)
|
||||
if value:
|
||||
is_valid, error = validator(value)
|
||||
if not is_valid:
|
||||
errors.append(f"{label}: {error}")
|
||||
|
||||
return len(errors) == 0, errors
|
||||
Reference in New Issue
Block a user