mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-27 15:17:18 +09:00
Juan DenisandClaude Fable 5
5417c39100
An audit found 102 FK columns with no index — every join, per-parent listing, and the delete cascades from _delete_cascade_policy scanned whole tables on them (deployments.app_id, audit_logs.user_id, dns_records.zone_id, webhook_logs.webhook_id, ...). All FK columns now declare index=True; migration 096 creates the same indexes on existing installs, deriving the list from the live schema at upgrade time (so it is idempotent and cannot drift from what an install actually has). Ratchet: test_every_fk_column_is_indexed pins the rule with no allowlist — a new FK column is born indexed or the build fails. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
97 lines
4.1 KiB
Python
97 lines
4.1 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, index=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}>'
|