193 lines
6.2 KiB
Python
Executable File
193 lines
6.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
VPN Access Server - Main Entry Point
|
|
|
|
Unified command-line interface for all VPN Access Server operations.
|
|
Supports authentication, session management, database operations, and testing.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
# Add the access module to the Python path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'access'))
|
|
|
|
def run_auth():
|
|
"""Run authentication module."""
|
|
from access.auth import main as auth_main
|
|
auth_main()
|
|
|
|
def run_session():
|
|
"""Run session management module."""
|
|
from access.session import main as session_main
|
|
session_main()
|
|
|
|
def run_init_db():
|
|
"""Initialize database schema."""
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'scripts'))
|
|
from scripts.init_db import main as init_db_main
|
|
init_db_main()
|
|
|
|
def run_seed_data():
|
|
"""Seed database with sample data."""
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'scripts'))
|
|
from scripts.seed_data import main as seed_data_main
|
|
seed_data_main()
|
|
|
|
def run_tests():
|
|
"""Run unit tests."""
|
|
import subprocess
|
|
try:
|
|
result = subprocess.run([
|
|
sys.executable, '-m', 'pytest', 'tests/', '-v'
|
|
], cwd=os.path.dirname(os.path.abspath(__file__)))
|
|
sys.exit(result.returncode)
|
|
except FileNotFoundError:
|
|
print("pytest not found. Install test dependencies with: uv add --group test pytest pytest-cov")
|
|
sys.exit(1)
|
|
|
|
def health_check():
|
|
"""Check system health and connectivity."""
|
|
from access.config import config
|
|
from access.db import db
|
|
from access.utils import setup_logging
|
|
|
|
logger = setup_logging(log_level='INFO')
|
|
|
|
print("🔍 VPN Access Server Health Check")
|
|
print("=" * 50)
|
|
|
|
# Configuration check
|
|
print("📋 Configuration...")
|
|
if config.validate():
|
|
print("✅ Configuration is valid")
|
|
else:
|
|
print("❌ Configuration validation failed")
|
|
return False
|
|
|
|
# Database check
|
|
print("🗄️ Database connectivity...")
|
|
if db.health_check():
|
|
print("✅ Database connection successful")
|
|
else:
|
|
print("❌ Database connection failed")
|
|
return False
|
|
|
|
# Test utilities
|
|
print("🔧 Utility functions...")
|
|
from access.utils import validate_mac_address, normalize_mac_address
|
|
test_mac = "00:11:22:33:44:55"
|
|
if validate_mac_address(test_mac) and normalize_mac_address("001122334455") == test_mac:
|
|
print("✅ Utility functions working")
|
|
else:
|
|
print("❌ Utility functions failed")
|
|
return False
|
|
|
|
print("=" * 50)
|
|
print("✅ All health checks passed!")
|
|
return True
|
|
|
|
def show_status():
|
|
"""Show system status and configuration."""
|
|
from access.config import config
|
|
|
|
print("📊 VPN Access Server Status")
|
|
print("=" * 50)
|
|
print(f"Database Host: {config.database.host}:{config.database.port}")
|
|
print(f"Database Name: {config.database.database}")
|
|
print(f"Database User: {config.database.username}")
|
|
print(f"Log Level: {config.server.log_level}")
|
|
print(f"Log File: {config.server.log_file}")
|
|
print(f"Default Session Limit: {config.server.default_session_limit}s ({config.server.default_session_limit//3600}h)")
|
|
print(f"Max Session Limit: {config.server.max_session_limit}s ({config.server.max_session_limit//3600}h)")
|
|
print("=" * 50)
|
|
|
|
from util.client.generate_client import generate_client_config
|
|
|
|
|
|
def run_gen_client(username: str):
|
|
"""Generate a client .ovpn file."""
|
|
generate_client_config(username)
|
|
|
|
|
|
def main():
|
|
"""Main entry point with command-line interface."""
|
|
parser = argparse.ArgumentParser(
|
|
description="VPN Access Server - OpenVPN authentication and session management",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
%(prog)s auth # Run authentication (for OpenVPN)
|
|
%(prog)s session # Run session management (for OpenVPN)
|
|
%(prog)s init-db # Initialize database schema
|
|
%(prog)s seed-data # Add sample data for testing
|
|
%(prog)s test # Run unit tests
|
|
%(prog)s health-check # Check system health
|
|
%(prog)s status # Show configuration status
|
|
%(prog)s gen-client <user> # Generate a client .ovpn file
|
|
|
|
Environment Variables:
|
|
DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD
|
|
LOG_LEVEL, DEFAULT_SESSION_LIMIT, MAX_SESSION_LIMIT
|
|
|
|
See .env.example for full configuration options.
|
|
"""
|
|
)
|
|
|
|
subparsers = parser.add_subparsers(dest='command', help='Commands')
|
|
|
|
subparsers.add_parser('auth', help='Run authentication (for OpenVPN)')
|
|
subparsers.add_parser('session', help='Run session management (for OpenVPN)')
|
|
subparsers.add_parser('init-db', help='Initialize database schema')
|
|
subparsers.add_parser('seed-data', help='Add sample data for testing')
|
|
subparsers.add_parser('test', help='Run unit tests')
|
|
subparsers.add_parser('health-check', help='Check system health')
|
|
subparsers.add_parser('status', help='Show configuration status')
|
|
|
|
gen_client_parser = subparsers.add_parser('gen-client', help='Generate a client .ovpn file')
|
|
gen_client_parser.add_argument('username', help='Username for the client config')
|
|
|
|
parser.add_argument(
|
|
'--version',
|
|
action='version',
|
|
version='VPN Access Server 1.0.0'
|
|
)
|
|
|
|
if len(sys.argv) < 2:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
if args.command == 'auth':
|
|
run_auth()
|
|
elif args.command == 'session':
|
|
run_session()
|
|
elif args.command == 'init-db':
|
|
run_init_db()
|
|
elif args.command == 'seed-data':
|
|
run_seed_data()
|
|
elif args.command == 'test':
|
|
run_tests()
|
|
elif args.command == 'health-check':
|
|
if not health_check():
|
|
sys.exit(1)
|
|
elif args.command == 'status':
|
|
show_status()
|
|
elif args.command == 'gen-client':
|
|
run_gen_client(args.username)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n⚠️ Operation cancelled by user")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|