diff --git a/app.py b/app.py index 436a491..f1cdc58 100644 --- a/app.py +++ b/app.py @@ -69,27 +69,29 @@ class FlicButtonHandler: raise def _decode_vapid_private_key(self): - """Load and validate VAPID private key with strict formatting.""" + """Load and strictly validate VAPID private key.""" try: + # Get and clean the key env_key = os.getenv('VAPID_PRIVATE_KEY', '').strip().strip('"\'') - - # Convert to consistent PEM format + + # Convert to clean PEM format if '\\n' in env_key: private_pem = env_key.replace('\\n', '\n') else: private_pem = env_key - + # Ensure proper PEM headers if not private_pem.startswith('-----BEGIN PRIVATE KEY-----'): private_pem = f"-----BEGIN PRIVATE KEY-----\n{private_pem}\n-----END PRIVATE KEY-----" - - # Validate by loading + + # Strict validation key = serialization.load_pem_private_key( private_pem.encode('utf-8'), - password=None + password=None, + backend=default_backend() ) - # Return standardized PEM + # Return in strict PEM format return key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, @@ -97,8 +99,8 @@ class FlicButtonHandler: ).decode('utf-8') except Exception as e: - logger.error(f"VAPID key loading failed: {str(e)}") - raise ValueError("Invalid VAPID private key format") from 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.""" @@ -120,50 +122,29 @@ class FlicButtonHandler: logger.error(f"Error saving subscriptions: {e}") async def send_push_notification(self, subscription: Dict, message: str): - """Send a web push notification with robust key handling.""" try: - if not self.subscriptions: - logger.warning("No subscriptions available") - return - - # Convert PEM key to bytes right before use - try: - private_key = serialization.load_pem_private_key( - self.vapid_private_key.encode('utf-8'), - password=None - ) - # Re-serialize to ensure clean format - vapid_private_key = private_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"Key conversion failed: {str(e)}") - raise - - # Extract domain for aud claim + # Get endpoint base for aud claim endpoint = subscription['endpoint'] - aud = endpoint.split('/fcm/send')[0] if 'fcm.googleapis.com' in endpoint else endpoint.split('/send')[0] + 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}") - logger.debug(f"Key length: {len(vapid_private_key)}") webpush( subscription_info=subscription, data=message, - vapid_private_key=vapid_private_key, + vapid_private_key=self.vapid_private_key, vapid_claims={ - "sub": os.getenv('VAPID_CLAIM_EMAIL', 'mailto:your-email@example.com'), - "aud": aud - } + "sub": os.getenv('VAPID_CLAIM_EMAIL'), + "aud": aud + "/" # Ensure trailing slash + }, + ttl=86400 # 24 hour expiration ) logger.info("Push notification sent successfully") - except WebPushException as e: + return True + except Exception as e: logger.error(f"Push failed: {str(e)}") - if 'Invalid JWT' in str(e): - logger.error("VAPID key validation failed") - raise + return False async def handle_button1(self): """Handle first button action - e.g., Home Lights On"""