mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 13:35:12 +09:00
Juan DenisandClaude Fable 5
4919ff1cfe
TimestampMixin (byte-identical to the hand-written pair — zero schema migrations) adopted by 52 classes across 44 model files; the utcnow tz-aware migration is now a one-file edit. uuid_pk() replaces the 13 pasted String(36) uuid4 lambdas across 7 files (ai.py's hex variant stays grandfathered, documented). Deviant shapes (indexed created_at, sanitization_profile's default-less updated_at, created-only models) deliberately keep their own columns. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
97 lines
4.0 KiB
Python
97 lines
4.0 KiB
Python
"""Organization-level chat / webhook connection for the Notification Bus.
|
|
|
|
A shared destination — a Discord/Slack/Telegram room or a generic webhook — that
|
|
receives notifications matching its category filter. Replaces the legacy global
|
|
``notifications.json`` chat config and the per-user webhook URL fields (which
|
|
become deprecated). Credentials are Fernet-encrypted per-field in
|
|
``credentials_json`` (never serialized); follows the ``EmailProviderConnection``
|
|
Connections pattern.
|
|
|
|
Per kind, ``credentials_json`` holds:
|
|
- discord / slack / webhook : {'url': ..., 'secret': <optional HMAC secret>}
|
|
- telegram : {'chat_id': ..., 'bot_token': ...}
|
|
"""
|
|
from datetime import datetime
|
|
|
|
from app import db
|
|
from app.models.mixins import TimestampMixin
|
|
from app.models.json_column_mixin import JsonColumnMixin
|
|
from app.utils.crypto import decrypt_secret_safe
|
|
|
|
|
|
class ChatWebhookConnection(TimestampMixin, JsonColumnMixin, db.Model):
|
|
__tablename__ = 'chat_webhook_connections'
|
|
|
|
KINDS = ('discord', 'slack', 'telegram', 'webhook')
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
kind = db.Column(db.String(20), nullable=False) # discord|slack|telegram|webhook
|
|
name = db.Column(db.String(120), nullable=False)
|
|
|
|
# Encrypted per-field credential map (see module docstring).
|
|
credentials_json = db.Column(db.Text)
|
|
|
|
# JSON list of categories this connection accepts (system/security/backups/
|
|
# apps). Null / empty = every category (a catch-all room).
|
|
categories_json = db.Column(db.Text)
|
|
|
|
is_active = db.Column(db.Boolean, default=True, nullable=False)
|
|
# Administrative default for this kind. Category fan-out remains driven by
|
|
# active connections whose category filters match the notification.
|
|
is_default = db.Column(db.Boolean, default=False, index=True)
|
|
# True when created by the one-time import of legacy notifications.json config.
|
|
imported = db.Column(db.Boolean, default=False)
|
|
|
|
created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
|
|
last_tested_at = db.Column(db.DateTime)
|
|
last_test_ok = db.Column(db.Boolean)
|
|
|
|
def raw_credentials(self):
|
|
# A hand-edited or corrupt row can hold valid JSON of the wrong shape
|
|
# (``[]``, a string, ``null``). Every caller assumes a mapping, so
|
|
# normalise here rather than at each call site -- same guard as
|
|
# ``categories()`` below.
|
|
return self._json_read('credentials_json', expect=dict)
|
|
|
|
def credentials(self):
|
|
"""Decrypted credentials for use at send/test time (never serialized)."""
|
|
out = {}
|
|
for key, value in self.raw_credentials().items():
|
|
out[key] = decrypt_secret_safe(value) if isinstance(value, str) else value
|
|
return out
|
|
|
|
def categories(self):
|
|
return self._json_read('categories_json', [], expect=list)
|
|
|
|
def matches_category(self, category):
|
|
"""A connection with no category filter accepts everything."""
|
|
cats = self.categories()
|
|
return not cats or category in cats
|
|
|
|
@property
|
|
def has_signing(self):
|
|
return bool(self.credentials().get('secret')) and self.kind == 'webhook'
|
|
|
|
def to_dict(self):
|
|
creds = self.credentials()
|
|
# Surface a masked destination hint, never the full URL / secret / token.
|
|
dest = creds.get('url') or creds.get('chat_id') or ''
|
|
masked = (dest[:24] + '…') if len(dest) > 25 else dest
|
|
return {
|
|
'id': self.id,
|
|
'kind': self.kind,
|
|
'name': self.name,
|
|
'destination': masked,
|
|
'categories': self.categories(),
|
|
'is_active': self.is_active,
|
|
'is_default': self.is_default,
|
|
'imported': self.imported,
|
|
'has_signing': self.has_signing,
|
|
'created_at': self.created_at.isoformat() if self.created_at else None,
|
|
'last_tested_at': self.last_tested_at.isoformat() if self.last_tested_at else None,
|
|
'last_test_ok': self.last_test_ok,
|
|
}
|
|
|
|
def __repr__(self):
|
|
return f'<ChatWebhookConnection {self.id} {self.kind} {self.name}>'
|