Initial setup for push notifications (still has issues with fcm aka chrome/opera)

Browse Source 
# Conflicts:
#	src/snek/templates/app.html

# Conflicts:
#	src/snek/app.py

# Conflicts:
#	src/snek/app.py
#	src/snek/templates/app.html

# Conflicts:
#	src/snek/app.py

# Conflicts:
#	src/snek/app.py
#	src/snek/static/push.js
#	src/snek/static/service-worker.js
#	src/snek/templates/app.html
This commit is contained in:
BordedDev 2025-03-17 03:16:39 +01:00
parent 4854d40508
commit 0057792802
No known key found for this signature in database
GPG Key ID: C5F495EAE56673BF
6 changed files with 330 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/snek/app.py
View File

@ -22,6 +22,8 @@ from app.app import Application as BaseApplication
from jinja2 import FileSystemLoader
from snek.sssh import start_ssh_server
from snek.system.notification import get_notifications
from snek.docs.app import Application as DocsApplication
from snek.mapper import get_mappers
from snek.service import get_services
@ -38,6 +40,7 @@ from snek.view.drive import DriveView
from snek.view.drive import DriveApiView
from snek.view.index import IndexView
from snek.view.login import LoginView
from snek.view.push import PushView
from snek.view.logout import LogoutView
from snek.view.register import RegisterView
from snek.view.rpc import RPCView
@ -101,6 +104,8 @@ class Application(BaseApplication):
self.time_start = datetime.now()
self.ssh_host = "0.0.0.0"
self.ssh_port = 2242
get_notifications()
self.setup_router()
self.ssh_server = None
self.sync_service = None
@ -208,6 +213,7 @@ class Application(BaseApplication):
self.router.add_view("/settings/index.html", SettingsIndexView)
self.router.add_view("/settings/profile.html", SettingsProfileView)
self.router.add_view("/settings/profile.json", SettingsProfileView)
self.router.add_view("/push.json", PushView)
self.router.add_view("/web.html", WebView)
self.router.add_view("/login.html", LoginView)
self.router.add_view("/login.json", LoginView)

82
src/snek/static/push.js
View File

@ -1,34 +1,54 @@
this.onpush = (event) => {
console.log(event.data);
// From here we can write the data to IndexedDB, send it to any open
// windows, display a notification, etc.
};
// this.onpush = (event) => {
// console.log(event.data);
// // From here we can write the data to IndexedDB, send it to any open
// // windows, display a notification, etc.
// };
function arrayBufferToBase64(buffer) {
return btoa(String.fromCharCode(...new Uint8Array(buffer)))
}
const keyResponse = await fetch('/push.json')
const keyData = await keyResponse.json()
const publicKey = Uint8Array.from(atob(keyData.publicKey), c => c.charCodeAt(0))
navigator.serviceWorker
.register("/service-worker.js")
.then((serviceWorkerRegistration) => {
serviceWorkerRegistration.pushManager.subscribe().then(
(pushSubscription) => {
const subscriptionObject = {
endpoint: pushSubscription.endpoint,
keys: {
p256dh: pushSubscription.getKey("p256dh"),
auth: pushSubscription.getKey("auth"),
},
encoding: PushManager.supportedContentEncodings,
/* other app-specific data, such as user identity */
};
console.log(
pushSubscription.endpoint,
pushSubscription,
subscriptionObject,
.register("/service-worker.js")
.then((serviceWorkerRegistration) => {
console.log(serviceWorkerRegistration);
serviceWorkerRegistration.pushManager.subscribe({
userVisibleOnly: true,
applicationServerKey: publicKey,
}).then(
(pushSubscription) => {
const subscriptionObject = {
...pushSubscription.toJSON(),
encoding: PushManager.supportedContentEncodings,
/* other app-specific data, such as user identity */
};
console.log(pushSubscription.endpoint, pushSubscription, pushSubscription.toJSON(), subscriptionObject);
// The push subscription details needed by the application
// server are now available, and can be sent to it using,
// for example, the fetch() API.
fetch('/push.json', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(subscriptionObject),
}).then((response) => {
if (!response.ok) {
throw new Error('Bad status code from server.');
}
return response.json();
}).then((responseData) => {
console.log(responseData);
});
},
(error) => {
console.error(error);
},
);
// The push subscription details needed by the application
// server are now available, and can be sent to it using,
// for example, the fetch() API.
},
(error) => {
console.error(error);
},
);
});
});

20
src/snek/static/service-worker.js
View File

@ -24,18 +24,14 @@ async function subscribeUser() {
}
// Service Worker (service-worker.js)
self.addEventListener("push", (event) => {
const data = event.data.json();
self.registration.showNotification(data.title, {
body: data.message,
icon: data.icon,
});
});
// self.addEventListener("push", (event) => {
// const data = event.data.json();
// self.registration.showNotification(data.title, {
// body: data.message,
// icon: data.icon,
// });
// });
/*
self.addEventListener("install", (event) => {
console.log("Service worker installed");
});
self.addEventListener("push", (event) => {
if (!(self.Notification && self.Notification.permission === "granted")) {
@ -62,4 +58,4 @@ self.addEventListener("notificationclick", (event) => {
event.notification.close();
event.waitUntil(clients.openWindow(
"https://snek.molodetz.nl",));
});*/
});

125
src/snek/system/notification.py Normal file
View File

@ -0,0 +1,125 @@
import time
import base64
import uuid
from functools import cache
import jwt
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
from urllib.parse import urlparse
import os.path
PRIVATE_KEY_FILE = './notification-private.pem'
PRIVATE_KEY_PKCS8_FILE = './notification-private.pkcs8.pem'
PUBLIC_KEY_FILE = './notification-public.pem'
def generate_private_key():
if not os.path.isfile(PRIVATE_KEY_FILE):
private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
# Serialize the private key to PEM format
pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
)
# Write the private key to a file
with open(PRIVATE_KEY_FILE, 'wb') as pem_out:
pem_out.write(pem)
def generate_pcks8_private_key():
# openssl pkcs8 -topk8 -nocrypt -in private.pem -out private.pkcs8.pem
if not os.path.isfile(PRIVATE_KEY_PKCS8_FILE):
with open(PRIVATE_KEY_FILE, 'rb') as pem_in:
private_key = serialization.load_pem_private_key(pem_in.read(), password=None, backend=default_backend())
# Serialize the private key to PKCS8 format
pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
# Write the private key to a file
with open(PRIVATE_KEY_PKCS8_FILE, 'wb') as pem_out:
pem_out.write(pem)
def generate_public_key():
if not os.path.isfile(PUBLIC_KEY_FILE):
with open(PRIVATE_KEY_FILE, 'rb') as pem_in:
private_key = serialization.load_pem_private_key(pem_in.read(), password=None, backend=default_backend())
# Get the public key from the private key
public_key = private_key.public_key()
# Serialize the public key to PEM format
pem = public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
# Write the public key to a file
with open(PUBLIC_KEY_FILE, 'wb') as pem_out:
pem_out.write(pem)
def ensure_certificates():
generate_private_key()
generate_pcks8_private_key()
generate_public_key()
class Notifications:
private_key_pem = None
public_key = None
public_key_jwk = None
def __init__(self):
ensure_certificates()
with open(PRIVATE_KEY_FILE, 'rb') as pem_in:
private_key = serialization.load_pem_private_key(pem_in.read(), password=None, backend=default_backend())
self.private_key_pem = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
)
with open(PUBLIC_KEY_FILE, 'rb') as pem_in:
self.public_key = serialization.load_pem_public_key(pem_in.read(), backend=default_backend())
public_numbers = self.public_key.public_numbers()
self.public_key_jwk = {
"kty": "EC",
"crv": "P-256",
"x": base64.urlsafe_b64encode(public_numbers.x.to_bytes(32, byteorder='big')).decode('utf-8'),
"y": base64.urlsafe_b64encode(public_numbers.y.to_bytes(32, byteorder='big')).decode('utf-8')
}
print(f"Public key JWK: {self.public_key_jwk}")
def create_notification_authorization(self, push_url):
target = urlparse(push_url)
aud = f"{target.scheme}://{target.netloc}"
sub = "mailto:admin@molodetz.nl"
identifier = str(uuid.uuid4())
print(f"Creating notification authorization for {aud} with identifier {identifier}")
return jwt.encode({
"sub": sub,
"aud": aud,
"exp": int(time.time()) + 60 * 60,
"nbf": int(time.time()),
"iat": int(time.time()),
"jti": identifier,
}, self.private_key_pem, algorithm='ES256')
@cache
def get_notifications():
return Notifications()

9
src/snek/templates/app.html
View File

@ -7,9 +7,7 @@
<title>Snek</title>
<style>{{highlight_styles}}</style>
<script src="/polyfills/Promise.withResolvers.js" type="module"></script>
<!--
<script src="/push.js"></script>
-->
<script src="/push.js" type="module"></script>
<script src="/fancy-button.js" type="module"></script>
<script src="/upload-button.js" type="module"></script>
<script src="/generic-form.js" type="module"></script>
@ -34,9 +32,6 @@
<script defer src="https://umami.molodetz.nl/script.js" data-website-id="d127c3e4-dc70-4041-a1c8-bcc32c2492ea"></script>
</head>
<body>
<header>
<div class="logo no-select">{% block header_text %}{% endblock %}</div>
<nav class="no-select" style="overflow:hidden;scroll-behavior:smooth">
@ -54,7 +49,7 @@
{% block sidebar %}
{% include "sidebar_channels.html" %}
{% endblock %}
<main>
<main>
{% block main %}
<chat-window class="chat-area"></chat-window>

124
src/snek/view/push.py Normal file
View File

@ -0,0 +1,124 @@
# Written by retoor@molodetz.nl
# This code defines an async class-based view called StatusView for handling HTTP GET requests. It fetches user details and their associated channel memberships from a database and returns a JSON response with user information if the user is logged in.
# The code uses an imported module `BaseView`. There are dependencies on the `snek.system.view` module which provides the BaseView class.
# MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
# PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
# CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE
# OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import base64
import requests
from cryptography.hazmat.primitives import serialization
from snek.system.notification import get_notifications
from snek.system.view import BaseFormView
class PushView(BaseFormView):
async def get(self):
notifications =get_notifications()
return await self.json_response({
"publicKey": base64.b64encode(
notifications.public_key.public_bytes(
encoding=serialization.Encoding.X962,
format=serialization.PublicFormat.UncompressedPoint
)
).decode('utf-8').rstrip("="),
})
async def post(self):
memberships = []
user = {}
user_id = self.session.get("uid")
if user_id:
user = await self.app.services.user.get(uid=user_id)
if not user:
return await self.json_response({"error": "User not found"}, status=404)
body = await self.request.json()
if not ("encoding" in body and "endpoint" in body and "keys" in body and "p256dh" in body["keys"] and "auth" in body["keys"]):
return await self.json_response({"error": "Invalid request"}, status=400)
print(body)
notifications =get_notifications()
cert = base64.b64encode(
notifications.public_key.public_bytes(
encoding=serialization.Encoding.X962,
format=serialization.PublicFormat.UncompressedPoint
)
).decode('utf-8').rstrip("=")
headers = {
"TTL": "60",
"Authorization": f"WebPush {notifications.create_notification_authorization(body['endpoint'])}",
"Crypto-Key": f"p256ecdsa={cert}",
}
print(headers)
post_notification = requests.post(
body["endpoint"],
headers=headers)
print(post_notification.status_code)
print(post_notification.text)
async for model in self.app.services.channel_member.find(
user_uid=user_id, deleted_at=None, is_banned=False
):
channel = await self.app.services.channel.get(uid=model["channel_uid"])
memberships.append(
{
"name": channel["label"],
"description": model["description"],
"user_uid": model["user_uid"],
"is_moderator": model["is_moderator"],
"is_read_only": model["is_read_only"],
"is_muted": model["is_muted"],
"is_banned": model["is_banned"],
"channel_uid": model["channel_uid"],
"uid": model["uid"],
}
)
user = {
"username": user["username"],
"email": user["email"],
"nick": user["nick"],
"uid": user["uid"],
"color": user["color"],
"memberships": memberships,
}
return await self.json_response(
{
"user": user,
"cache": await self.app.cache.create_cache_key(
self.app.cache.cache, None
),
}
)