rewritten

.gitignore

@@ -0,0 +1,4 @@
+myenv
+.vscode
+subscriptions.json
+.env

Dockerfile

@@ -0,0 +1,26 @@
+FROM python:3.11-slim
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements file
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application files
+COPY . .
+
+# Generate VAPID keys if .env doesn't exist
+RUN if [ ! -f .env ]; then python generate_vapid_keys.py; fi
+
+# Expose the application port
+EXPOSE 8080
+
+# Command to run the application
+CMD ["python", "app.py"]

README.md

@@ -1,3 +1,133 @@
-# flic-webpush-localstack
+# Flic Button Web Push Notification Service
 
-AWS Lambda function using LocalStack in Docker that receives HTTP requests from Flic buttons and sends WebPush notifications.
+## Overview
+This application provides a dockerized solution for handling Flic smart button events and sending web push notifications to a Progressive Web App (PWA).
+
+## Features
+- Webhook endpoint for Flic button events
+- Web Push notification support
+- Configurable button actions
+- Subscription management
+
+## Prerequisites
+- Docker
+- Docker Compose
+- Traefik network
+- Curl or Postman for testing
+
+## Setup
+
+### 1. Generate VAPID Keys
+Run the VAPID key generation script:
+```bash
+python generate_vapid_keys.py
+```
+This will create a `.env` file with VAPID keys.
+
+### 2. Configure Flic Buttons
+Edit the `.env` file to add your Flic button serial numbers:
+```
+FLIC_BUTTON1_SERIAL=your_button1_serial
+FLIC_BUTTON2_SERIAL=your_button2_serial
+FLIC_BUTTON3_SERIAL=your_button3_serial
+```
+
+### 3. Docker Compose Configuration
+```yaml
+version: '3'
+services:
+  flic-webpush:
+    build: .
+    volumes:
+      - ./subscriptions.json:/app/subscriptions.json
+    networks:
+      - traefik
+    labels:
+      - "traefik.enable=true"
+      - "traefik.http.routers.flic-webpush.rule=Host(`flic.yourdomain.com`)"
+
+networks:
+  traefik:
+    external: true
+```
+
+### 4. Endpoints
+- `/flic-webhook`: Receive Flic button events
+- `/subscribe`: Add web push subscriptions
+
+## Testing Webhooks
+
+### Simulating Flic Button Events
+You can test the webhook endpoint using curl or Postman. Here are example requests:
+
+#### Button 1 Event (Home Lights On)
+```bash
+curl -X POST http://localhost:8080/flic-webhook \
+     -H "Content-Type: application/json" \
+     -d '{
+         "serial": "your_button1_serial",
+         "event": "click",
+         "timestamp": "'$(date -u +"%Y-%m-%dT%H:%M:%SZ")'"
+     }'
+```
+
+#### Button 2 Event (Security System Arm)
+```bash
+curl -X POST http://localhost:8080/flic-webhook \
+     -H "Content-Type: application/json" \
+     -d '{
+         "serial": "your_button2_serial",
+         "event": "double_click",
+         "timestamp": "'$(date -u +"%Y-%m-%dT%H:%M:%SZ")'"
+     }'
+```
+
+#### Button 3 Event (Panic Alert)
+```bash
+curl -X POST http://localhost:8080/flic-webhook \
+     -H "Content-Type: application/json" \
+     -d '{
+         "serial": "your_button3_serial",
+         "event": "long_press",
+         "timestamp": "'$(date -u +"%Y-%m-%dT%H:%M:%SZ")'"
+     }'
+```
+
+### Adding a Web Push Subscription
+To test the subscription endpoint:
+```bash
+curl -X POST http://localhost:8080/subscribe \
+     -H "Content-Type: application/json" \
+     -d '{
+         "endpoint": "https://example.com/push-endpoint",
+         "keys": {
+             "p256dh": "base64-public-key",
+             "auth": "base64-auth-secret"
+         }
+     }'
+```
+
+### Debugging Tips
+- Check container logs: `docker logs flic-webpush`
+- Verify subscription file: `cat subscriptions.json`
+- Ensure correct button serial numbers in `.env`
+
+## Button Actions
+- Button 1: Home Lights On
+- Button 2: Security System Arm
+- Button 3: Panic Alert
+
+## Logging
+Configurable via `LOG_LEVEL` in `.env`
+
+## Security Considerations
+- Keep VAPID keys secret
+- Use HTTPS
+- Validate and sanitize all incoming webhook requests
+- Implement proper authentication for production use
+
+## Troubleshooting
+- Ensure all environment variables are correctly set
+- Check network connectivity
+- Verify Traefik configuration
+- Validate button serial numbers match between configuration and webhook

app.py

@@ -0,0 +1,289 @@
+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())

generate_vapid_keys.py

@@ -0,0 +1,101 @@
+#!/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()

requirements.txt

@@ -0,0 +1,4 @@
+aiohttp==3.9.3
+pywebpush==2.0.0
+python-dotenv==1.0.0
+cryptography==42.0.4