2HoursProject/flic-webhook-webpush
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first version

Browse Source
This commit is contained in:
cpu
2025-03-26 03:09:49 +01:00
parent d14e7a4446
commit 95a5b893ec
6 changed files with 495 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored Normal file
View File

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

26
Dockerfile Normal file
View File

@@ -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"]

134
README.md
View File

@@ -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

233
app.py Normal file
View File

@@ -0,0 +1,233 @@
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.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__)
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):
"""
Decode and load the VAPID private key from base64 encoded string.
Returns the PEM-formatted private key as a string.
"""
try:
# Decode base64 private key
private_key_pem = base64.urlsafe_b64decode(
os.getenv('VAPID_PRIVATE_KEY', '').encode('utf-8')
)
# Load private key to validate it
private_key = serialization.load_pem_private_key(
private_key_pem,
password=None
)
# Return the original PEM string (pywebpush needs this format)
return private_key_pem.decode('utf-8')
except Exception as e:
logger.error(f"Error loading VAPID private key: {e}")
raise
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):
"""Send a web push notification."""
try:
if not self.subscriptions:
logger.warning("No subscriptions available")
return
webpush(
subscription_info=subscription,
data=message,
vapid_private_key=self.vapid_private_key,
vapid_claims={"sub": "mailto:your-email@example.com"}
)
except WebPushException as e:
logger.error(f"Push notification error: {e}")
# Remove invalid subscription
self.subscriptions = [s for s in self.subscriptions if s != subscription]
self.save_subscriptions()
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."""
if not self.subscriptions:
logger.warning("No subscriptions to broadcast to")
return
tasks = [
self.send_push_notification(subscription, message)
for subscription in self.subscriptions
]
await asyncio.gather(*tasks)
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()
# Setup 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())

96
generate_vapid_keys.py Normal file
View File

@@ -0,0 +1,96 @@
#!/usr/bin/env python3
import os
import base64
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
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())
# Serialize private key
private_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
# Get public key
public_key = private_key.public_key()
public_pem = public_key.public_bytes(
encoding=serialization.Encoding.X962,
format=serialization.PublicFormat.UncompressedPoint
)
# Base64 encode keys
env_vars['VAPID_PRIVATE_KEY'] = base64.urlsafe_b64encode(private_pem).decode('utf-8')
env_vars['VAPID_PUBLIC_KEY'] = base64.urlsafe_b64encode(public_pem).decode('utf-8')
print("New VAPID keys generated and added to .env file.")
else:
print("Existing VAPID keys found in .env file - 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', '/app/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()

4
requirements.txt Normal file
View File

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