From f614e67e8b13372d310856ebb611bf510db915f3 Mon Sep 17 00:00:00 2001 From: cpu Date: Mon, 31 Mar 2025 23:03:23 +0200 Subject: [PATCH] clen up --- app.py | 289 ----------------------------------------- generate_vapid_keys.py | 101 -------------- server.js | 3 - 3 files changed, 393 deletions(-) delete mode 100644 app.py delete mode 100644 generate_vapid_keys.py diff --git a/app.py b/app.py deleted file mode 100644 index 379d5ab..0000000 --- a/app.py +++ /dev/null @@ -1,289 +0,0 @@ -import asyncio -import json -import logging -import os -import base64 -from typing import Dict, List -import signal -import pathlib - -import aiohttp -from aiohttp import web -from dotenv import load_dotenv -from pywebpush import webpush, WebPushException -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import ec - -# Load environment variables -load_dotenv() - -# Configure logging -logging.basicConfig( - level=getattr(logging, os.getenv('LOG_LEVEL', 'INFO')), - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) -logger = logging.getLogger(__name__) - -# CORS Configuration -ALLOWED_ORIGINS = [ - "https://game-timer.virtonline.eu", - # Add other allowed origins if needed -] -ALLOWED_METHODS = ["POST", "OPTIONS"] -ALLOWED_HEADERS = ["Content-Type"] - -class FlicButtonHandler: - def __init__(self): - # Load button configurations - self.button_configs = { - os.getenv('FLIC_BUTTON1_SERIAL'): self.handle_button1, - os.getenv('FLIC_BUTTON2_SERIAL'): self.handle_button2, - os.getenv('FLIC_BUTTON3_SERIAL'): self.handle_button3 - } - - # Ensure subscriptions file and directory exist - self.subscriptions_file = os.getenv('SUBSCRIPTIONS_FILE', 'app/subscriptions.json') - self._ensure_subscriptions_file() - - # Load subscriptions - self.subscriptions = self.load_subscriptions() - - # Prepare VAPID keys - self.vapid_private_key = self._decode_vapid_private_key() - - def _ensure_subscriptions_file(self): - """ - Ensure the subscriptions file and its parent directory exist. - Create them if they don't. - """ - try: - # Create parent directory if it doesn't exist - pathlib.Path(self.subscriptions_file).parent.mkdir(parents=True, exist_ok=True) - - # Create file if it doesn't exist - if not os.path.exists(self.subscriptions_file): - with open(self.subscriptions_file, 'w') as f: - json.dump([], f) - except Exception as e: - logger.error(f"Error ensuring subscriptions file: {e}") - raise - - def _decode_vapid_private_key(self): - """Load and strictly validate VAPID private key.""" - try: - # Get and clean the key - env_key = os.getenv('VAPID_PRIVATE_KEY', '').strip() - - # Reconstruct PEM format if missing headers - if not env_key.startswith('-----BEGIN PRIVATE KEY-----'): - env_key = f"-----BEGIN PRIVATE KEY-----\n{env_key}\n-----END PRIVATE KEY-----" - - # Strict validation and key preparation - key = serialization.load_pem_private_key( - env_key.encode('utf-8'), - password=None, - backend=default_backend() - ) - - # Return in strict PEM format - return key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption() - ).decode('utf-8') - - except Exception as e: - logger.error(f"CRITICAL: Invalid VAPID key - {str(e)}") - raise RuntimeError("Invalid VAPID private key configuration") from e - - def load_subscriptions(self) -> List[Dict]: - """Load web push subscriptions from file.""" - try: - with open(self.subscriptions_file, 'r') as f: - # Handle empty file case - content = f.read().strip() - return json.loads(content) if content else [] - except json.JSONDecodeError: - logger.error(f"Error decoding subscriptions from {self.subscriptions_file}") - return [] - - def save_subscriptions(self): - """Save web push subscriptions to file.""" - try: - with open(self.subscriptions_file, 'w') as f: - json.dump(self.subscriptions, f, indent=2) - except Exception as e: - logger.error(f"Error saving subscriptions: {e}") - - async def send_push_notification(self, subscription: Dict, message: str): - try: - # Determine audience (aud) claim for VAPID - endpoint = subscription['endpoint'] - aud = (endpoint.split('/send')[0] if '/send' in endpoint - else endpoint.split('/fcm/send')[0]) - - logger.debug(f"Sending to: {endpoint[:50]}...") - logger.debug(f"Using aud: {aud}") - - # Perform web push - result = webpush( - subscription_info=subscription, - data=message, - vapid_private_key=self.vapid_private_key, - vapid_claims={ - "sub": os.getenv('VAPID_CLAIM_EMAIL'), - "aud": aud + "/" # Ensure trailing slash - }, - ttl=86400 # 24 hour expiration - ) - logger.info("Push notification sent successfully") - return True - except Exception as e: - logger.error(f"Push failed: {str(e)}") - logger.error(f"Endpoint details: {subscription['endpoint']}") - logger.error(f"Keys: {subscription.get('keys', 'No keys found')}") - return False - - async def handle_button1(self): - """Handle first button action - e.g., Home Lights On""" - logger.info("Button 1 pressed: Home Lights On") - message = json.dumps({"action": "home_lights_on"}) - await self.broadcast_notification(message) - - async def handle_button2(self): - """Handle second button action - e.g., Security System Arm""" - logger.info("Button 2 pressed: Security System Arm") - message = json.dumps({"action": "security_arm"}) - await self.broadcast_notification(message) - - async def handle_button3(self): - """Handle third button action - e.g., Panic Button""" - logger.info("Button 3 pressed: Panic Alert") - message = json.dumps({"action": "panic_alert"}) - await self.broadcast_notification(message) - - async def broadcast_notification(self, message: str): - """Broadcast notification to all subscriptions with error handling.""" - if not self.subscriptions: - logger.warning("No subscriptions to broadcast to") - return - - success_count = 0 - for subscription in self.subscriptions: - try: - success = await self.send_push_notification(subscription, message) - if success: - success_count += 1 - except Exception as e: - logger.error(f"Failed to send to {subscription['endpoint'][:30]}...: {str(e)}") - - logger.info(f"Notifications sent: {success_count}/{len(self.subscriptions)}") - - async def handle_flic_webhook(self, request): - """Webhook endpoint for Flic button events.""" - try: - data = await request.json() - button_serial = data.get('serial') - - # Validate button serial - if button_serial not in self.button_configs: - logger.warning(f"Unknown button serial: {button_serial}") - return web.Response(status=400) - - # Call the corresponding button handler - handler = self.button_configs[button_serial] - await handler() - - return web.Response(status=200) - except Exception as e: - logger.error(f"Error processing Flic webhook: {e}") - return web.Response(status=500) - - async def handle_subscribe(self, request): - """Add a new web push subscription.""" - try: - subscription = await request.json() - - # Check if subscription already exists - if subscription not in self.subscriptions: - self.subscriptions.append(subscription) - self.save_subscriptions() - logger.info("New subscription added") - - return web.Response(status=200) - except Exception as e: - logger.error(f"Subscription error: {e}") - return web.Response(status=500) - -def create_app(): - """Create and configure the aiohttp application.""" - app = web.Application() - handler = FlicButtonHandler() - - async def options_handler(request): - """Handle OPTIONS requests for CORS preflight.""" - origin = request.headers.get('Origin', '') - if origin in ALLOWED_ORIGINS: - headers = { - 'Access-Control-Allow-Origin': origin, - 'Access-Control-Allow-Methods': ', '.join(ALLOWED_METHODS), - 'Access-Control-Allow-Headers': ', '.join(ALLOWED_HEADERS), - 'Access-Control-Max-Age': '86400', # 24 hours - } - return web.Response(status=200, headers=headers) - return web.Response(status=403) # Forbidden origin - - async def add_cors_headers(request, response): - """Add CORS headers to normal responses.""" - origin = request.headers.get('Origin', '') - if origin in ALLOWED_ORIGINS: - response.headers['Access-Control-Allow-Origin'] = origin - response.headers['Access-Control-Expose-Headers'] = 'Content-Type' - return response - - # Register middleware - app.on_response_prepare.append(add_cors_headers) - - # Setup routes with OPTIONS handlers - app.router.add_route('OPTIONS', '/flic-webhook', options_handler) - app.router.add_route('OPTIONS', '/subscribe', options_handler) - - # Original routes - app.router.add_post('/flic-webhook', handler.handle_flic_webhook) - app.router.add_post('/subscribe', handler.handle_subscribe) - - return app - -async def main(): - """Main application entry point.""" - app = create_app() - runner = web.AppRunner(app) - await runner.setup() - site = web.TCPSite(runner, '0.0.0.0', 8080) - await site.start() - - logger.info("Application started on port 8080") - - # Create an event to keep the application running - stop_event = asyncio.Event() - - def signal_handler(): - """Handle interrupt signals to gracefully stop the application.""" - logger.info("Received shutdown signal") - stop_event.set() - - # Register signal handlers - loop = asyncio.get_running_loop() - for sig in (signal.SIGINT, signal.SIGTERM): - loop.add_signal_handler(sig, signal_handler) - - # Wait until stop event is set - await stop_event.wait() - - # Cleanup - await runner.cleanup() - logger.info("Application shutting down") - -if __name__ == '__main__': - asyncio.run(main()) diff --git a/generate_vapid_keys.py b/generate_vapid_keys.py deleted file mode 100644 index b0d01cf..0000000 --- a/generate_vapid_keys.py +++ /dev/null @@ -1,101 +0,0 @@ -#!/usr/bin/env python3 -import os -import base64 -from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.backends import default_backend - -def generate_vapid_keys(): - """ - Generate VAPID keys and update .env file, preserving existing variables. - Only regenerates keys if they are missing or empty. - """ - # Read existing .env file if it exists - env_vars = {} - if os.path.exists('.env'): - with open('.env', 'r') as f: - for line in f: - line = line.strip() - if line and not line.startswith('#') and '=' in line: - key, value = line.split('=', 1) - env_vars[key.strip()] = value.strip() - - # Check if we need to generate new keys - need_new_keys = ( - 'VAPID_PRIVATE_KEY' not in env_vars or - 'VAPID_PUBLIC_KEY' not in env_vars or - not env_vars.get('VAPID_PRIVATE_KEY') or - not env_vars.get('VAPID_PUBLIC_KEY') - ) - - if need_new_keys: - # Generate EC private key - private_key = ec.generate_private_key(ec.SECP256R1(), backend=default_backend()) - - # Serialize private key to PEM format, but keep it clean - private_pem = private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption() - ).decode('utf-8') - - # Clean up PEM formatting for .env file - private_pem_clean = private_pem.replace('-----BEGIN PRIVATE KEY-----\n', '').replace('\n-----END PRIVATE KEY-----\n', '').replace('\n', '') - - # Get public key - public_key = private_key.public_key() - public_key_bytes = public_key.public_bytes( - encoding=serialization.Encoding.X962, - format=serialization.PublicFormat.UncompressedPoint - ) - - # Store keys - env_vars['VAPID_PRIVATE_KEY'] = private_pem_clean - env_vars['VAPID_PUBLIC_KEY'] = base64.urlsafe_b64encode(public_key_bytes).decode('utf-8') - - print("New VAPID keys generated in .env-compatible format.") - else: - print("Existing VAPID keys found - no changes made.") - - # Ensure we have all required configuration variables with defaults if missing - defaults = { - # Flic Button Configuration - 'FLIC_BUTTON1_SERIAL': env_vars.get('FLIC_BUTTON1_SERIAL', 'your_button1_serial'), - 'FLIC_BUTTON2_SERIAL': env_vars.get('FLIC_BUTTON2_SERIAL', 'your_button2_serial'), - 'FLIC_BUTTON3_SERIAL': env_vars.get('FLIC_BUTTON3_SERIAL', 'your_button3_serial'), - - # Subscription Storage - 'SUBSCRIPTIONS_FILE': env_vars.get('SUBSCRIPTIONS_FILE', 'data/subscriptions.json'), - - # Logging Configuration - 'LOG_LEVEL': env_vars.get('LOG_LEVEL', 'INFO'), - - # VAPID Claim (email) - 'VAPID_CLAIM_EMAIL': env_vars.get('VAPID_CLAIM_EMAIL', 'mailto:your-email@example.com') - } - - # Update env_vars with defaults for any missing keys - env_vars.update({k: v for k, v in defaults.items() if k not in env_vars}) - - # Write back to .env file - with open('.env', 'w') as f: - f.write("# VAPID Keys for Web Push\n") - f.write(f"VAPID_PRIVATE_KEY={env_vars['VAPID_PRIVATE_KEY']}\n") - f.write(f"VAPID_PUBLIC_KEY={env_vars['VAPID_PUBLIC_KEY']}\n\n") - - f.write("# Flic Button Configuration\n") - f.write(f"FLIC_BUTTON1_SERIAL={env_vars['FLIC_BUTTON1_SERIAL']}\n") - f.write(f"FLIC_BUTTON2_SERIAL={env_vars['FLIC_BUTTON2_SERIAL']}\n") - f.write(f"FLIC_BUTTON3_SERIAL={env_vars['FLIC_BUTTON3_SERIAL']}\n\n") - - f.write("# Subscription Storage\n") - f.write(f"SUBSCRIPTIONS_FILE={env_vars['SUBSCRIPTIONS_FILE']}\n\n") - - f.write("# Logging Configuration\n") - f.write(f"LOG_LEVEL={env_vars['LOG_LEVEL']}\n\n") - - f.write("# VAPID Claim Email\n") - f.write(f"VAPID_CLAIM_EMAIL={env_vars['VAPID_CLAIM_EMAIL']}\n") - -if __name__ == '__main__': - generate_vapid_keys() diff --git a/server.js b/server.js index 16ba3e9..fc24ffe 100644 --- a/server.js +++ b/server.js @@ -313,7 +313,6 @@ app.get('/webhook/:click_type', authenticateBasic, async (req, res) => { batteryLevel: batteryLevel // Use the extracted value } // icon: '/path/to/icon.png' - // icon: '/path/to/icon.png' }); try { @@ -368,8 +367,6 @@ const closeGracefully = (signal) => { process.exit(0); }); - // Force close server after 10 seconds - // Force close server after 10 seconds setTimeout(() => { logger.error('Could not close connections in time, forcefully shutting down');