File main.py of Package PersistenceOS
#!/usr/bin/env python3
'''
FILE : main.py
PROJECT : PersistenceOS
COPYRIGHT : (c) 2024 PersistenceOS Team
AUTHOR : PersistenceOS Team
PACKAGE : PersistenceOS
LICENSE : MIT
PURPOSE : Main FastAPI application entry point
'''
import os
import sys
import json
import logging
# Add user site-packages to Python path (for pip --user installations)
import site
user_site = site.getusersitepackages()
if user_site and os.path.exists(user_site):
sys.path.insert(0, user_site)
# Try importing required packages with helpful error messages
try:
import uvicorn
from fastapi import FastAPI, Depends, HTTPException, status, Request, Form
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.responses import JSONResponse, FileResponse, HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from typing import Optional, Dict, List, Any
from datetime import datetime, timedelta
import subprocess
import socket
import platform
import psutil
from fastapi.templating import Jinja2Templates
except ImportError as e:
print(f"Error importing required packages: {e}")
print("Please ensure FastAPI, Uvicorn, and psutil are installed:")
print("Try: pip3.11 install --user fastapi uvicorn psutil")
print("Or: pip3 install --user fastapi uvicorn psutil")
sys.exit(1)
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("persistenceos.api")
# Environment variables
API_PORT = int(os.environ.get("API_PORT", 8080))
DEBUG_MODE = os.environ.get("DEBUG_MODE", "false").lower() == "true"
WEB_ROOT = os.environ.get("WEB_ROOT", "/usr/lib/persistence/web-ui")
API_CONFIG_PATH = os.environ.get("API_CONFIG_PATH", "/usr/lib/persistence/web-ui/api-config.json")
# Add this function after the logger initialization
def check_web_files():
"""Check and log all available web UI files at startup."""
logger.info("Checking web UI files...")
paths_to_check = [
WEB_ROOT,
"/var/lib/persistence/web-ui",
"/usr/lib/persistence/web-ui"
]
for path in paths_to_check:
if os.path.exists(path):
logger.info(f"Found web UI directory: {path}")
try:
# Count files in directory
file_count = sum(1 for _ in os.walk(path))
logger.info(f"Directory {path} contains {file_count} files/directories")
# Check for login.html specifically
login_path = os.path.join(path, "login.html")
if os.path.exists(login_path):
logger.info(f"Found login.html at {login_path} ({os.path.getsize(login_path)} bytes)")
else:
logger.warning(f"login.html not found at {login_path}")
# Check static directory
static_path = os.path.join(path, "static")
if os.path.exists(static_path) and os.path.isdir(static_path):
logger.info(f"Found static directory at {static_path}")
static_files = []
for root, dirs, files in os.walk(static_path):
for file in files:
static_files.append(os.path.join(root, file))
logger.info(f"Static directory contains {len(static_files)} files")
else:
logger.warning(f"Static directory not found at {static_path}")
# Log a few root files
try:
root_files = [f for f in os.listdir(path) if os.path.isfile(os.path.join(path, f))]
logger.info(f"Root files in {path}: {', '.join(root_files[:5])}{'...' if len(root_files) > 5 else ''}")
except Exception as e:
logger.error(f"Error listing files in {path}: {str(e)}")
except Exception as e:
logger.error(f"Error examining {path}: {str(e)}")
else:
logger.warning(f"Web UI directory not found: {path}")
logger.info("Web UI files check complete")
# Create FastAPI app
app = FastAPI(
title="PersistenceOS API",
description="API for PersistenceOS management interface",
version="6.1.0",
docs_url="/api/docs" if DEBUG_MODE else None,
redoc_url="/api/redoc" if DEBUG_MODE else None,
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Check web files at startup
check_web_files()
# OAuth2 security scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token")
# =============================================
# User authentication and token handling
# =============================================
# In-memory token store (for development; in production use proper storage)
active_tokens = {}
# Mock user database (in production, this would come from PAM or system users)
def get_user(username: str) -> Optional[dict]:
"""Get user by username from system users or mock DB in development mode."""
if DEBUG_MODE:
# Development mode - use mock users
users = {
"root": {
"username": "root",
"display_name": "Administrator",
"roles": ["admin"],
"email": "admin@persistenceos.org",
"password": "linux" # only used in development
}
}
return users.get(username)
else:
# Production mode - validate against system users
try:
# Simple check if user exists on system
result = subprocess.run(
["getent", "passwd", username],
capture_output=True, text=True, check=False
)
if result.returncode == 0:
# User exists, create basic user object
return {
"username": username,
"display_name": username,
"roles": ["admin"] if username == "root" else ["user"]
}
except Exception as e:
logger.error(f"Error checking system user: {e}")
return None
def authenticate_user(username: str, password: str) -> Optional[dict]:
"""Authenticate user against system or mock DB in development mode."""
user = get_user(username)
if not user:
return None
if DEBUG_MODE:
# In development mode, check against mock password
if user.get("password") == password:
return user
else:
# In production, validate against system authentication
try:
# This is a mock implementation - would use PAM in real system
# For security reasons we don't implement actual auth here
# but use mock successful auth for root/linux
if username == "root" and password == "linux":
return user
except Exception as e:
logger.error(f"Authentication error: {e}")
return None
def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
"""Create a new access token."""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=60)
to_encode.update({"exp": expire})
token = f"dev-token-{os.urandom(8).hex()}" # Mock token
# Store token with expiration
active_tokens[token] = {
"data": to_encode,
"expires_at": int(expire.timestamp() * 1000) # milliseconds
}
return token
async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
"""Get the current user from a token."""
if token not in active_tokens:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token or token expired",
headers={"WWW-Authenticate": "Bearer"},
)
token_data = active_tokens[token]
# Check if token is expired
now = datetime.utcnow()
if now.timestamp() * 1000 > token_data["expires_at"]:
# Remove expired token
del active_tokens[token]
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
headers={"WWW-Authenticate": "Bearer"},
)
user_data = token_data["data"].get("sub")
user = get_user(user_data)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found",
headers={"WWW-Authenticate": "Bearer"},
)
return user
async def get_current_user_optional(token: str) -> Optional[dict]:
"""Get current user from token without raising exceptions."""
try:
if not token:
return None
if token not in active_tokens:
return None
token_data = active_tokens[token]
# Check if token is expired
now = datetime.utcnow()
if now.timestamp() * 1000 > token_data["expires_at"]:
# Remove expired token
del active_tokens[token]
return None
user_data = token_data["data"].get("sub")
user = get_user(user_data)
return user
except Exception as e:
logger.error(f"Error validating token: {e}")
return None
# =============================================
# API Routes
# =============================================
# Configure templates with the correct directory
templates_dir = os.path.join(WEB_ROOT)
templates = Jinja2Templates(directory=templates_dir)
logger.info(f"Templates directory configured: {templates_dir}")
# Static files will be mounted later in configure_static_files() to avoid conflicts
# with specific route handlers
logger.info("Static file mounting deferred to configure_static_files() function")
@app.get("/")
async def root():
"""Redirect to login page."""
logger.info(f"Accessing root path, redirecting to login")
# Redirect to /login for the login page
return RedirectResponse(url="/login")
@app.get("/login")
async def serve_login(request: Request):
"""Serve the bulletproof login page"""
logger.info("Serving bulletproof login page")
# Try to serve bulletproof-login.js first
bulletproof_login_paths = [
os.path.join(WEB_ROOT, "js/bulletproof-login.js"),
os.path.join(WEB_ROOT, "static/js/bulletproof-login.js"),
"/usr/lib/persistence/web-ui/js/bulletproof-login.js"
]
for login_path in bulletproof_login_paths:
if os.path.exists(login_path):
logger.info(f"Serving bulletproof login from: {login_path}")
return FileResponse(login_path, media_type="application/javascript")
# Fallback to legacy login.html if bulletproof not available
legacy_paths = [
os.path.join(WEB_ROOT, "static/login.html"),
os.path.join(WEB_ROOT, "login.html"),
"/usr/lib/persistence/web-ui/static/login.html",
"/usr/lib/persistence/web-ui/login.html"
]
for login_path in legacy_paths:
if os.path.exists(login_path):
logger.info(f"Serving legacy login page from: {login_path}")
return FileResponse(login_path, media_type="text/html")
# Final fallback to legacy JS login
logger.warning("No login files found, falling back to legacy JS login")
return RedirectResponse(url="/js/login.js", status_code=302)
@app.get("/index", response_class=HTMLResponse)
async def serve_index(request: Request, current_user: dict = Depends(get_current_user)):
"""Protected dashboard route"""
index_path = os.path.join(WEB_ROOT, "index.html")
logger.info(f"Serving index from: {index_path}")
if os.path.exists(index_path):
try:
return templates.TemplateResponse(
"index.html",
{
"request": request,
"user": current_user
}
)
except Exception as e:
logger.error(f"Error rendering index template: {str(e)}")
# Fallback to direct file response if templating fails
return FileResponse(
index_path,
media_type="text/html"
)
else:
logger.error(f"index.html not found at {index_path}")
raise HTTPException(status_code=404, detail=f"index.html not found")
@app.get("/index.html", response_class=HTMLResponse)
async def serve_index_html(request: Request, current_user: dict = Depends(get_current_user)):
"""Alternative route for index.html"""
return await serve_index(request, current_user)
@app.get("/api/debug/files")
async def debug_files():
"""Debug endpoint to list files in the web root."""
result = {
"web_root": WEB_ROOT,
"exists": os.path.exists(WEB_ROOT),
"files": [],
"subdirs": []
}
if os.path.exists(WEB_ROOT):
# List direct files
for item in os.listdir(WEB_ROOT):
item_path = os.path.join(WEB_ROOT, item)
if os.path.isfile(item_path):
result["files"].append({
"name": item,
"size": os.path.getsize(item_path),
"readable": os.access(item_path, os.R_OK),
"modified": datetime.fromtimestamp(os.path.getmtime(item_path)).isoformat()
})
elif os.path.isdir(item_path):
subdir = {"name": item, "files": []}
# List files in subdirectory
try:
for subitem in os.listdir(item_path):
subitem_path = os.path.join(item_path, subitem)
if os.path.isfile(subitem_path):
subdir["files"].append({
"name": subitem,
"size": os.path.getsize(subitem_path),
"readable": os.access(subitem_path, os.R_OK)
})
except Exception as e:
subdir["error"] = str(e)
result["subdirs"].append(subdir)
return result
@app.get("/api/debug/js-files")
async def debug_js_files():
"""Debug endpoint specifically for JavaScript files."""
js_dir = os.path.join(WEB_ROOT, "js")
result = {
"web_root": WEB_ROOT,
"js_directory": js_dir,
"exists": os.path.exists(js_dir),
"files": [],
"current_working_directory": os.getcwd(),
"absolute_js_dir": os.path.abspath(js_dir)
}
if os.path.exists(js_dir):
try:
for item in os.listdir(js_dir):
item_path = os.path.join(js_dir, item)
if os.path.isfile(item_path):
result["files"].append({
"name": item,
"path": item_path,
"absolute_path": os.path.abspath(item_path),
"size": os.path.getsize(item_path),
"readable": os.access(item_path, os.R_OK),
"modified": datetime.fromtimestamp(os.path.getmtime(item_path)).isoformat()
})
except Exception as e:
result["error"] = str(e)
# Also check specific files we're looking for
specific_files = ["app.js", "auth.js", "vue.js"]
result["specific_file_checks"] = {}
for filename in specific_files:
file_path = os.path.join(js_dir, filename)
result["specific_file_checks"][filename] = {
"path": file_path,
"exists": os.path.exists(file_path),
"absolute_path": os.path.abspath(file_path),
"readable": os.access(file_path, os.R_OK) if os.path.exists(file_path) else False
}
return result
@app.get("/api/debug/webui-path")
async def debug_webui_path(request: Request):
"""Debug endpoint to check web UI path and file structure."""
logger.info("Debug webui-path endpoint accessed")
files = []
for root, _, filenames in os.walk(WEB_ROOT):
for filename in filenames:
full_path = os.path.join(root, filename)
rel_path = os.path.relpath(full_path, WEB_ROOT)
files.append({
"path": rel_path,
"size": os.path.getsize(full_path),
"readable": os.access(full_path, os.R_OK),
"executable": os.access(full_path, os.X_OK),
"modified": os.path.getmtime(full_path)
})
# Also include root_path information from FastAPI
return {
"webui_path": WEB_ROOT,
"exists": os.path.exists(WEB_ROOT),
"file_count": len(files),
"files": files,
"root_path": request.scope.get("root_path", ""),
}
@app.get("/api/debug/fallback")
async def debug_fallback():
"""Debug fallback route handler."""
logger.error(f"File not found in fallback route: {{full_path}}")
return JSONResponse(
status_code=404,
content={"detail": f"{{path}} not found at {{full_path}}"}
)
# Protected endpoint example
@app.get("/api/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
if not current_user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authentication required",
headers={"WWW-Authenticate": "Bearer"},
)
return {
"message": f"Hello, {current_user['username']}! This is a protected endpoint.",
"user": current_user
}
@app.get("/api/health")
async def health_check():
"""Health check endpoint."""
return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
@app.get("/api/config")
async def get_config():
"""Get server configuration."""
try:
# Get primary IP address
hostname = socket.gethostname()
primary_ip = socket.gethostbyname(hostname)
# Get all non-loopback IPv4 addresses
available_ips = []
for interface, addrs in psutil.net_if_addrs().items():
for addr in addrs:
if addr.family == socket.AF_INET and not addr.address.startswith('127.'):
available_ips.append(addr.address)
# Create config object
config = {
"host": primary_ip,
"port": API_PORT,
"http_port": API_PORT,
"https_port": 8443, # Default HTTPS port
"api_base_url": f"http://{primary_ip}:{API_PORT}/api",
"secure_api_base_url": f"https://{primary_ip}:8443/api",
"available_ips": available_ips,
"features": {
"ssl_enabled": os.path.exists("/etc/ssl/private/persistenceos.key"),
"api_enabled": True,
"debug_enabled": DEBUG_MODE
},
"system": {
"version": "6.1.0",
"hostname": hostname,
"platform": platform.platform(),
"generated_at": datetime.utcnow().isoformat()
}
}
# Save to api-config.json for web UI
with open(API_CONFIG_PATH, 'w') as f:
json.dump(config, f, indent=2)
return config
except Exception as e:
logger.error(f"Error generating config: {e}")
return JSONResponse(
status_code=500,
content={"error": f"Failed to generate configuration: {str(e)}"}
)
@app.get("/api-config.json")
async def api_config_json():
"""Return the API configuration for web UI."""
try:
# Check if file exists, otherwise generate it
if not os.path.exists(API_CONFIG_PATH):
await get_config()
# Return the file
return FileResponse(API_CONFIG_PATH)
except Exception as e:
logger.error(f"Error serving API config: {e}")
return JSONResponse(
status_code=500,
content={"error": f"Failed to serve API configuration: {str(e)}"}
)
@app.post("/api/auth/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
"""
OAuth2 compatible token login, get an access token for future requests.
This endpoint follows the OAuth2 password flow standard.
"""
user = authenticate_user(form_data.username, form_data.password)
if not user:
logger.warning(f"Failed login attempt for user: {form_data.username}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
logger.info(f"Successful login for user: {form_data.username}")
# Create access token with 1 hour expiry
access_token_expires = timedelta(hours=1)
access_token = create_access_token(
data={"sub": user["username"]},
expires_delta=access_token_expires
)
# Get token expiry timestamp for frontend compatibility
token_data = active_tokens[access_token]
# Return the token in the format expected by OAuth2 and our frontend
return {
"access_token": access_token,
"token_type": "bearer",
"expires_in": access_token_expires.total_seconds(),
"expires_at": token_data["expires_at"], # Add expires_at for frontend
"user": {
"username": user["username"],
"display_name": user.get("display_name", user["username"]),
"roles": user.get("roles", []),
"email": user.get("email", "")
}
}
@app.post("/api/auth/refresh")
async def refresh_token(current_user: dict = Depends(get_current_user)):
"""Refresh access token."""
# Create new token with 1 hour expiry
token_expires = timedelta(hours=1)
token = create_access_token(
data={"sub": current_user["username"]},
expires_delta=token_expires
)
# Get token expiry timestamp
token_data = active_tokens[token]
return {
"access_token": token,
"token_type": "bearer",
"expires_at": token_data["expires_at"]
}
@app.post("/login")
async def handle_login(
request: Request,
username: str = Form(...),
password: str = Form(...)
):
"""Handle form submission"""
# Replace with your actual auth logic
if username == "root" and password == "linux":
token = create_access_token(
data={"sub": username},
expires_delta=timedelta(hours=1)
)
response = RedirectResponse(url="/app.html", status_code=303)
response.set_cookie(
key="access_token",
value=f"Bearer {token}",
httponly=True,
secure=True
)
return response
# Try to find login.html template for error rendering
possible_paths = [
os.path.join(WEB_ROOT, "login.html"),
"/var/lib/persistence/web-ui/login.html",
"/usr/lib/persistence/web-ui/login.html"
]
found_path = None
for path in possible_paths:
if os.path.exists(path):
found_path = path
break
# If template exists, try to render with error
if found_path:
try:
return templates.TemplateResponse(
"login.html",
{
"request": request,
"version": "6.1.0",
"error": "Invalid credentials"
},
status_code=401
)
except Exception as e:
logger.error(f"Error rendering login template with error: {str(e)}")
# Fallback HTML response with error
html_content = f"""<!DOCTYPE html>
<html>
<head>
<title>PersistenceOS Login</title>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<style>
body {{ font-family: Arial, sans-serif; display: flex; justify-content: center; align-items: center; height: 100vh; margin: 0; }}
.login-container {{ width: 300px; padding: 20px; box-shadow: 0 0 10px rgba(0,0,0,0.1); }}
h1 {{ text-align: center; color: #333; }}
input {{ width: 100%; padding: 10px; margin: 10px 0; box-sizing: border-box; }}
button {{ width: 100%; padding: 10px; background-color: #4CAF50; color: white; border: none; cursor: pointer; }}
.error {{ color: red; margin-top: 10px; text-align: center; }}
</style>
</head>
<body>
<div class="login-container">
<h1>PersistenceOS</h1>
<div class="error">Invalid credentials</div>
<form id="login-form" method="post" action="/api/auth/token">
<input type="text" name="username" placeholder="Username" autocomplete="username" required>
<input type="password" name="password" placeholder="Password" autocomplete="current-password" required>
<button type="submit">Login</button>
</form>
</div>
<script>
document.getElementById('login-form').addEventListener('submit', async (e) => {{
e.preventDefault();
const formData = new FormData(e.target);
try {{
const response = await fetch('/api/auth/token', {{
method: 'POST',
body: new URLSearchParams(formData),
headers: {{ 'Content-Type': 'application/x-www-form-urlencoded' }}
}});
if (response.ok) {{
window.location.href = '/app.html';
}} else {{
document.getElementById('error').textContent = 'Login failed. Check credentials.';
}}
}} catch (err) {{
document.getElementById('error').textContent = 'Connection error. Please try again.';
}}
}});
</script>
</body>
</html>"""
return HTMLResponse(content=html_content, status_code=401)
# =============================================
# System Information Endpoints
# =============================================
@app.get("/api/system/info")
async def system_info(current_user: dict = Depends(get_current_user)):
"""Get system information."""
try:
# Get basic system information
hostname = socket.gethostname()
kernel = platform.release()
arch = platform.machine()
# Get uptime
uptime_seconds = int(time.time() - psutil.boot_time())
days, remainder = divmod(uptime_seconds, 86400)
hours, remainder = divmod(remainder, 3600)
minutes, seconds = divmod(remainder, 60)
uptime = f"{days} days, {hours} hours, {minutes} minutes"
# Get CPU info
cpu_count = psutil.cpu_count(logical=True)
cpu_usage = psutil.cpu_percent(interval=0.1)
# Get memory info
memory = psutil.virtual_memory()
memory_total = memory.total
memory_used = memory.used
memory_percent = memory.percent
# Get disk info
disk = psutil.disk_usage('/')
disk_total = disk.total
disk_used = disk.used
disk_percent = disk.percent
return {
"hostname": hostname,
"kernel": kernel,
"architecture": arch,
"uptime": uptime,
"uptime_seconds": uptime_seconds,
"cpu": {
"count": cpu_count,
"usage_percent": cpu_usage
},
"memory": {
"total": memory_total,
"used": memory_used,
"percent": memory_percent
},
"disk": {
"total": disk_total,
"used": disk_used,
"percent": disk_percent
}
}
except Exception as e:
logger.error(f"Error getting system info: {e}")
return JSONResponse(
status_code=500,
content={"error": f"Failed to get system information: {str(e)}"}
)
@app.get("/api/system/network/interfaces")
async def network_interfaces(current_user: dict = Depends(get_current_user)):
"""Get network interface information."""
try:
interfaces = []
# Get network interfaces
for interface, addrs in psutil.net_if_addrs().items():
# Skip loopback interfaces
if interface.startswith('lo'):
continue
# Get interface stats
stats = psutil.net_if_stats().get(interface, None)
is_up = stats.isup if stats else False
# Get IPv4 address
ipv4 = None
for addr in addrs:
if addr.family == socket.AF_INET:
ipv4 = addr.address
break
# Only include interfaces with IPv4 addresses
if ipv4:
interfaces.append({
"name": interface,
"ip": ipv4,
"status": "up" if is_up else "down",
"mac": next((addr.address for addr in addrs if addr.family == psutil.AF_LINK), "")
})
return interfaces
except Exception as e:
logger.error(f"Error getting network interfaces: {e}")
return JSONResponse(
status_code=500,
content={"error": f"Failed to get network interfaces: {str(e)}"}
)
# =============================================
# Static files and startup
# =============================================
def configure_static_files():
"""Configure static file serving."""
if not os.path.exists(WEB_ROOT):
logger.error(f"Web root directory does not exist: {WEB_ROOT}")
return False
try:
# List web root contents for debugging
logger.info(f"Web root directory contents {WEB_ROOT}:")
for item in os.listdir(WEB_ROOT):
item_path = os.path.join(WEB_ROOT, item)
if os.path.isfile(item_path):
logger.info(f" File: {item} ({os.path.getsize(item_path)} bytes)")
else:
logger.info(f" Directory: {item}")
# Check login.html specifically
login_path = os.path.join(WEB_ROOT, "login.html")
if os.path.exists(login_path):
logger.info(f"login.html exists at {login_path} ({os.path.getsize(login_path)} bytes)")
else:
logger.error(f"login.html does not exist at {login_path}")
# Mount static files after explicit routes
static_path = os.path.join(WEB_ROOT, "static")
if os.path.exists(static_path):
app.mount("/static", StaticFiles(directory=static_path), name="static")
logger.info(f"Mounted /static directory from {static_path}")
else:
logger.warning(f"Static directory not found at {static_path}")
# DO NOT mount the entire web root at "/" as it conflicts with our specific routes
# The specific route handlers will serve the files instead
logger.info("Static file configuration complete - using specific route handlers for JS/CSS files")
return True
except Exception as e:
logger.error(f"Failed to configure static files: {e}")
return False
def start():
"""Start the API server."""
# Configure static files
configure_static_files()
# Start uvicorn
logger.info(f"Starting PersistenceOS API on port {API_PORT}")
uvicorn.run(
"main:app",
host="0.0.0.0",
port=API_PORT,
log_level="info",
reload=DEBUG_MODE
)
# Updated handler for direct /login.html requests - redirect to new login system
@app.get("/login.html", response_class=HTMLResponse)
async def serve_login_html(request: Request):
"""Direct handler for /login.html - redirect to new login system"""
logger.info("Legacy /login.html request, redirecting to /login")
return RedirectResponse(url="/login", status_code=301) # Permanent redirect
@app.get("/app.html", response_class=HTMLResponse)
async def serve_app_html(request: Request):
"""Serve Vue.js app HTML wrapper"""
logger.info("Serving app.html - authentication will be handled by client-side JavaScript")
app_html = f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>PersistenceOS - Dashboard</title>
<link rel="stylesheet" href="/css/style.css">
<link rel="icon" href="/img/favicon.ico" type="image/x-icon">
<!-- Font Awesome for icons -->
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.4.0/css/all.min.css">
<!-- Authentication library -->
<script src="/js/auth.js"></script>
<!-- Vue.js (with CDN fallback) -->
<script>
// Try to load local Vue.js first, fallback to CDN
const vueScript = document.createElement('script');
vueScript.src = '/js/vue.js';
vueScript.onerror = function() {{
console.log('Local Vue.js not found, loading from CDN...');
const cdnScript = document.createElement('script');
cdnScript.src = 'https://unpkg.com/vue@3/dist/vue.global.prod.js';
cdnScript.onload = function() {{
console.log('Vue.js loaded from CDN successfully');
}};
cdnScript.onerror = function() {{
console.error('Failed to load Vue.js from CDN');
document.getElementById('app').innerHTML = '<div style="text-align: center; padding: 50px; color: red;">Error: Could not load Vue.js framework</div>';
}};
document.head.appendChild(cdnScript);
}};
vueScript.onload = function() {{
console.log('Vue.js loaded locally successfully');
}};
document.head.appendChild(vueScript);
</script>
<!-- Main app script -->
<script src="/js/app.js" defer></script>
</head>
<body>
<div id="app">
<!-- Vue.js app will be mounted here -->
<div class="loading-indicator">Loading PersistenceOS Dashboard...</div>
</div>
</body>
</html>
"""
return HTMLResponse(content=app_html, headers={{
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}})
@app.get("/js/app.js")
async def serve_app_js(request: Request):
"""Serve the Vue.js application script"""
app_js_path = os.path.join(WEB_ROOT, "js/app.js")
logger.info(f"Attempting to serve app.js from: {app_js_path}")
logger.info(f"WEB_ROOT is: {WEB_ROOT}")
logger.info(f"File exists: {os.path.exists(app_js_path)}")
# Try multiple possible locations
possible_locations = [
app_js_path,
os.path.join("/usr/lib/persistence/web-ui/js", "app.js"),
os.path.join("/usr/lib/persistence/web-ui/static/js", "app.js"),
"app.js" # Current directory
]
for location in possible_locations:
logger.info(f"Checking app.js at: {location} - exists: {os.path.exists(location)}")
if os.path.exists(location):
logger.info(f"Found app.js at: {location}")
return FileResponse(location, media_type="application/javascript")
# Enhanced debugging
logger.error(f"app.js not found in any location: {possible_locations}")
# List directory contents for debugging
for check_dir in [WEB_ROOT, "/usr/lib/persistence/web-ui", "/usr/lib/persistence/web-ui/js"]:
if os.path.exists(check_dir):
try:
contents = os.listdir(check_dir)
logger.info(f"Contents of {check_dir}: {contents}")
except Exception as e:
logger.error(f"Error listing {check_dir}: {e}")
else:
logger.error(f"Directory not found: {check_dir}")
raise HTTPException(status_code=404, detail=f"app.js not found in any expected location")
@app.get("/js/auth.js")
async def serve_auth_js(request: Request):
"""Serve the authentication script"""
auth_js_path = os.path.join(WEB_ROOT, "js/auth.js")
logger.info(f"Attempting to serve auth.js from: {auth_js_path}")
logger.info(f"WEB_ROOT is: {WEB_ROOT}")
logger.info(f"File exists: {os.path.exists(auth_js_path)}")
# Try multiple possible locations
possible_locations = [
auth_js_path,
os.path.join("/usr/lib/persistence/web-ui/js", "auth.js"),
os.path.join("/usr/lib/persistence/web-ui/static/js", "auth.js"),
"auth.js" # Current directory
]
for location in possible_locations:
logger.info(f"Checking auth.js at: {location} - exists: {os.path.exists(location)}")
if os.path.exists(location):
logger.info(f"Found auth.js at: {location}")
return FileResponse(location, media_type="application/javascript")
# Enhanced debugging
logger.error(f"auth.js not found in any location: {possible_locations}")
raise HTTPException(status_code=404, detail=f"auth.js not found in any expected location")
@app.get("/js/vue.js")
async def serve_vue_js(request: Request):
"""Serve the Vue.js library"""
vue_js_path = os.path.join(WEB_ROOT, "js/vue.js")
logger.info(f"Attempting to serve vue.js from: {vue_js_path}")
logger.info(f"File exists: {os.path.exists(vue_js_path)}")
if os.path.exists(vue_js_path):
return FileResponse(vue_js_path, media_type="application/javascript")
else:
logger.error(f"vue.js not found at {vue_js_path}")
# List directory contents for debugging
js_dir = os.path.join(WEB_ROOT, "js")
if os.path.exists(js_dir):
logger.info(f"Contents of {js_dir}: {os.listdir(js_dir)}")
else:
logger.error(f"JS directory not found at {js_dir}")
raise HTTPException(status_code=404, detail=f"vue.js not found at {vue_js_path}")
@app.get("/js/login.js")
async def serve_login_js(request: Request):
"""Serve the login script (legacy fallback)"""
login_js_path = os.path.join(WEB_ROOT, "js/login.js")
logger.info(f"Attempting to serve login.js from: {login_js_path}")
# Try multiple possible locations
possible_locations = [
login_js_path,
os.path.join("/usr/lib/persistence/web-ui/js", "login.js"),
os.path.join("/usr/lib/persistence/web-ui/static/js", "login.js"),
"login.js" # Current directory
]
for location in possible_locations:
if os.path.exists(location):
logger.info(f"Found login.js at: {location}")
return FileResponse(location, media_type="application/javascript")
logger.error(f"login.js not found in any location: {possible_locations}")
raise HTTPException(status_code=404, detail=f"login.js not found in any expected location")
@app.get("/js/bulletproof-login.js")
async def serve_bulletproof_login_js(request: Request):
"""Serve the bulletproof login script"""
bulletproof_login_paths = [
os.path.join(WEB_ROOT, "js/bulletproof-login.js"),
os.path.join(WEB_ROOT, "static/js/bulletproof-login.js"),
"/usr/lib/persistence/web-ui/js/bulletproof-login.js"
]
for location in bulletproof_login_paths:
if os.path.exists(location):
logger.info(f"Found bulletproof-login.js at: {location}")
return FileResponse(location, media_type="application/javascript")
logger.error(f"bulletproof-login.js not found in any location: {bulletproof_login_paths}")
raise HTTPException(status_code=404, detail=f"bulletproof-login.js not found")
@app.get("/js/unified-auth.js")
async def serve_unified_auth_js(request: Request):
"""Serve the unified authentication script"""
unified_auth_paths = [
os.path.join(WEB_ROOT, "js/unified-auth.js"),
os.path.join(WEB_ROOT, "static/js/unified-auth.js"),
"/usr/lib/persistence/web-ui/js/unified-auth.js"
]
for location in unified_auth_paths:
if os.path.exists(location):
logger.info(f"Found unified-auth.js at: {location}")
return FileResponse(location, media_type="application/javascript")
logger.error(f"unified-auth.js not found in any location: {unified_auth_paths}")
raise HTTPException(status_code=404, detail=f"unified-auth.js not found")
@app.get("/js/bulletproof-app.js")
async def serve_bulletproof_app_js(request: Request):
"""Serve the bulletproof app script"""
bulletproof_app_paths = [
os.path.join(WEB_ROOT, "js/bulletproof-app.js"),
os.path.join(WEB_ROOT, "static/js/bulletproof-app.js"),
"/usr/lib/persistence/web-ui/js/bulletproof-app.js"
]
for location in bulletproof_app_paths:
if os.path.exists(location):
logger.info(f"Found bulletproof-app.js at: {location}")
return FileResponse(location, media_type="application/javascript")
logger.error(f"bulletproof-app.js not found in any location: {bulletproof_app_paths}")
raise HTTPException(status_code=404, detail=f"bulletproof-app.js not found")
@app.get("/js/vue.global.prod.js")
async def serve_vue_global_prod_js(request: Request):
"""Serve the Vue.js library (legacy route for compatibility)"""
# Redirect to the new vue.js route
logger.info("Legacy vue.global.prod.js route accessed, redirecting to vue.js")
return await serve_vue_js(request)
@app.get("/css/style.css")
async def serve_style_css(request: Request):
"""Serve the main stylesheet"""
css_path = os.path.join(WEB_ROOT, "css/style.css")
if os.path.exists(css_path):
return FileResponse(css_path, media_type="text/css")
else:
logger.error(f"style.css not found at {css_path}")
raise HTTPException(status_code=404, detail=f"style.css not found")
@app.get("/img/favicon.ico")
async def serve_favicon(request: Request):
"""Serve the favicon"""
favicon_path = os.path.join(WEB_ROOT, "img/favicon.ico")
if os.path.exists(favicon_path):
return FileResponse(favicon_path, media_type="image/x-icon")
else:
logger.error(f"favicon.ico not found at {favicon_path}")
raise HTTPException(status_code=404, detail=f"favicon.ico not found")
@app.get("/static/js/login.js")
async def serve_static_login_js(request: Request):
"""Serve the login.js from static directory"""
login_js_path = os.path.join(WEB_ROOT, "static/js/login.js")
logger.info(f"Attempting to serve static login.js from: {login_js_path}")
if os.path.exists(login_js_path):
return FileResponse(login_js_path, media_type="application/javascript")
else:
logger.error(f"static login.js not found at {login_js_path}")
# Try to find login.js in other locations
alternative_paths = [
os.path.join(WEB_ROOT, "js/login.js"),
"login.js"
]
for alt_path in alternative_paths:
if os.path.exists(alt_path):
logger.info(f"Found login.js at alternative location: {alt_path}")
return FileResponse(alt_path, media_type="application/javascript")
raise HTTPException(status_code=404, detail=f"login.js not found")
# Catch-all route for any other JavaScript files
@app.get("/js/{filename}")
async def serve_js_files(filename: str, request: Request):
"""Serve any JavaScript file from the web UI directory"""
logger.info(f"Generic JS route called for: {filename}")
logger.info(f"WEB_ROOT is: {WEB_ROOT}")
# Try multiple possible locations
possible_locations = [
os.path.join(WEB_ROOT, "js", filename),
os.path.join("/usr/lib/persistence/web-ui/js", filename),
os.path.join("/usr/lib/persistence/web-ui/static/js", filename),
filename # Current directory
]
for location in possible_locations:
logger.info(f"Checking {filename} at: {location} - exists: {os.path.exists(location)}")
if os.path.exists(location):
logger.info(f"Found {filename} at: {location}")
return FileResponse(location, media_type="application/javascript")
# Enhanced debugging
logger.error(f"{filename} not found in any location: {possible_locations}")
# List directory contents for debugging
for check_dir in [WEB_ROOT, "/usr/lib/persistence/web-ui", "/usr/lib/persistence/web-ui/js"]:
if os.path.exists(check_dir):
try:
contents = os.listdir(check_dir)
logger.info(f"Contents of {check_dir}: {contents}")
except Exception as e:
logger.error(f"Error listing {check_dir}: {e}")
else:
logger.error(f"Directory not found: {check_dir}")
raise HTTPException(status_code=404, detail=f"{filename} not found in any expected location")
if __name__ == "__main__":
import time # Import needed for the uptime calculation
import sys
# Handle command line arguments
if len(sys.argv) > 1 and sys.argv[1] == "--debug":
logger.info("Debug mode enabled via command line")
# Override DEBUG_MODE environment variable
import os
os.environ["DEBUG_MODE"] = "true"
# Update the global DEBUG_MODE variable
globals()["DEBUG_MODE"] = True
# Generate initial configuration
try:
logger.info("Generating initial API configuration...")
with app.router.stall_next_request():
get_config()
except Exception as e:
logger.error(f"Error generating initial configuration: {e}")
# Start the application
start()
