feat(observability): add notification and audit services
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
// Package audit stores the deliberately small, redacted security audit trail.
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"database/sql"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Event struct {
|
||||
ID string `json:"id"`
|
||||
ActorID string `json:"actor_id"`
|
||||
ActorLabel string `json:"actor_label"`
|
||||
InstanceID string `json:"instance_id"`
|
||||
Action string `json:"action"`
|
||||
Outcome string `json:"outcome"`
|
||||
OccurredAt time.Time `json:"occurred_at"`
|
||||
Summary map[string]string `json:"summary"`
|
||||
}
|
||||
|
||||
type Filter struct {
|
||||
ActorID, InstanceID, Action, Outcome string
|
||||
Since, Until time.Time
|
||||
Limit int
|
||||
}
|
||||
|
||||
type Policy struct {
|
||||
RetentionDays int `json:"retention_days"`
|
||||
MaximumCount int `json:"maximum_count"`
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
db *sql.DB
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func New(db *sql.DB) *Service { return &Service{db: db, now: time.Now} }
|
||||
|
||||
var allowedSummaryKeys = map[string]bool{"target_id": true, "target_name": true, "channel_type": true, "event_type": true, "reason_code": true, "deleted_count": true, "before": true, "after": true}
|
||||
|
||||
func (s *Service) Record(ctx context.Context, event Event) error {
|
||||
if event.Action == "" || (event.Outcome != "allowed" && event.Outcome != "denied" && event.Outcome != "failed") {
|
||||
return errors.New("invalid audit event")
|
||||
}
|
||||
clean := map[string]string{}
|
||||
for key, value := range event.Summary {
|
||||
if allowedSummaryKeys[key] && len(value) <= 200 {
|
||||
clean[key] = value
|
||||
}
|
||||
}
|
||||
body, _ := json.Marshal(clean)
|
||||
when := event.OccurredAt
|
||||
if when.IsZero() {
|
||||
when = s.now().UTC()
|
||||
}
|
||||
var actor, instance any
|
||||
if event.ActorID != "" {
|
||||
actor = event.ActorID
|
||||
}
|
||||
if event.InstanceID != "" {
|
||||
instance = event.InstanceID
|
||||
}
|
||||
_, err := s.db.ExecContext(ctx, `INSERT INTO audit_events(id,occurred_at,actor_id,actor_label,instance_id,action,outcome,summary_json) VALUES(?,?,?,?,?,?,?,?)`, randomID(), when.Format(time.RFC3339Nano), actor, event.ActorLabel, instance, event.Action, event.Outcome, string(body))
|
||||
if err != nil {
|
||||
return fmt.Errorf("record audit event: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) List(ctx context.Context, f Filter) ([]Event, error) {
|
||||
limit := f.Limit
|
||||
if limit <= 0 || limit > 200 {
|
||||
limit = 100
|
||||
}
|
||||
clauses, args := []string{"1=1"}, []any{}
|
||||
for _, item := range []struct{ column, value string }{{"actor_id", f.ActorID}, {"instance_id", f.InstanceID}, {"action", f.Action}, {"outcome", f.Outcome}} {
|
||||
if item.value != "" {
|
||||
clauses = append(clauses, item.column+"=?")
|
||||
args = append(args, item.value)
|
||||
}
|
||||
}
|
||||
if !f.Since.IsZero() {
|
||||
clauses = append(clauses, "occurred_at>=?")
|
||||
args = append(args, f.Since.UTC().Format(time.RFC3339Nano))
|
||||
}
|
||||
if !f.Until.IsZero() {
|
||||
clauses = append(clauses, "occurred_at<?")
|
||||
args = append(args, f.Until.UTC().Format(time.RFC3339Nano))
|
||||
}
|
||||
args = append(args, limit)
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT id,occurred_at,COALESCE(actor_id,''),actor_label,COALESCE(instance_id,''),action,outcome,summary_json FROM audit_events WHERE `+strings.Join(clauses, " AND ")+` ORDER BY occurred_at DESC,id DESC LIMIT ?`, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list audit events: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var events []Event
|
||||
for rows.Next() {
|
||||
var e Event
|
||||
var occurred, body string
|
||||
if err := rows.Scan(&e.ID, &occurred, &e.ActorID, &e.ActorLabel, &e.InstanceID, &e.Action, &e.Outcome, &body); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
e.OccurredAt, _ = time.Parse(time.RFC3339Nano, occurred)
|
||||
_ = json.Unmarshal([]byte(body), &e.Summary)
|
||||
events = append(events, e)
|
||||
}
|
||||
return events, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Service) Policy(ctx context.Context) (Policy, error) {
|
||||
var body string
|
||||
err := s.db.QueryRowContext(ctx, `SELECT value_json FROM system_settings WHERE key='audit_policy'`).Scan(&body)
|
||||
if err != nil {
|
||||
return Policy{}, err
|
||||
}
|
||||
var p Policy
|
||||
var raw struct {
|
||||
RetentionDays int `json:"retention_days"`
|
||||
MaximumCount int `json:"maximum_count"`
|
||||
}
|
||||
if err = json.Unmarshal([]byte(body), &raw); err != nil {
|
||||
return p, err
|
||||
}
|
||||
p.RetentionDays, p.MaximumCount = raw.RetentionDays, raw.MaximumCount
|
||||
return p, nil
|
||||
}
|
||||
func (s *Service) SetPolicy(ctx context.Context, p Policy) error {
|
||||
if p.RetentionDays < 0 || p.RetentionDays > 3650 || p.MaximumCount < 0 || p.MaximumCount > 1000000 {
|
||||
return errors.New("audit policy is out of bounds")
|
||||
}
|
||||
body, _ := json.Marshal(map[string]int{"retention_days": p.RetentionDays, "maximum_count": p.MaximumCount})
|
||||
_, err := s.db.ExecContext(ctx, `UPDATE system_settings SET value_json=?,revision=revision+1,updated_at=? WHERE key='audit_policy'`, body, s.now().UTC().Format(time.RFC3339Nano))
|
||||
return err
|
||||
}
|
||||
func (s *Service) Purge(ctx context.Context, before time.Time) (int64, error) {
|
||||
if before.IsZero() || before.After(s.now().UTC()) {
|
||||
return 0, errors.New("invalid audit purge boundary")
|
||||
}
|
||||
result, err := s.db.ExecContext(ctx, `DELETE FROM audit_events WHERE occurred_at < ?`, before.UTC().Format(time.RFC3339Nano))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return result.RowsAffected()
|
||||
}
|
||||
func (s *Service) RunRetention(ctx context.Context) (int64, error) {
|
||||
p, err := s.Policy(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var total int64
|
||||
if p.RetentionDays > 0 {
|
||||
n, e := s.Purge(ctx, s.now().UTC().AddDate(0, 0, -p.RetentionDays))
|
||||
if e != nil {
|
||||
return 0, e
|
||||
}
|
||||
total += n
|
||||
}
|
||||
if p.MaximumCount > 0 {
|
||||
result, e := s.db.ExecContext(ctx, `DELETE FROM audit_events WHERE id IN (SELECT id FROM audit_events ORDER BY occurred_at DESC,id DESC LIMIT -1 OFFSET ?)`, p.MaximumCount)
|
||||
if e != nil {
|
||||
return total, e
|
||||
}
|
||||
n, _ := result.RowsAffected()
|
||||
total += n
|
||||
}
|
||||
return total, nil
|
||||
}
|
||||
func randomID() string {
|
||||
b := make([]byte, 18)
|
||||
_, _ = rand.Read(b)
|
||||
return base64.RawURLEncoding.EncodeToString(b)
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package audit_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.zaynet.fr/DoGaMa/DoGaMa-serv/internal/audit"
|
||||
"git.zaynet.fr/DoGaMa/DoGaMa-serv/internal/persistence/sqlite"
|
||||
)
|
||||
|
||||
func TestRecordFiltersSummaryAndRetention(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
db, err := sqlite.Open(ctx, filepath.Join(t.TempDir(), "dogama.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
service := audit.New(db)
|
||||
old := time.Now().UTC().AddDate(0, 0, -40)
|
||||
if err := service.Record(ctx, audit.Event{OccurredAt: old, ActorLabel: "admin", Action: "instance.update", Outcome: "allowed", Summary: map[string]string{"target_name": "Palworld", "secret": "must-not-persist"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
events, err := service.List(ctx, audit.Filter{Action: "instance.update"})
|
||||
if err != nil || len(events) != 1 {
|
||||
t.Fatalf("events=%#v err=%v", events, err)
|
||||
}
|
||||
if events[0].Summary["target_name"] != "Palworld" || events[0].Summary["secret"] != "" {
|
||||
t.Fatalf("summary was not allow-listed: %#v", events[0].Summary)
|
||||
}
|
||||
if err := service.SetPolicy(ctx, audit.Policy{RetentionDays: 30, MaximumCount: 100}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
deleted, err := service.RunRetention(ctx)
|
||||
if err != nil || deleted != 1 {
|
||||
t.Fatalf("deleted=%d err=%v", deleted, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPolicyAndPurgeBounds(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
db, err := sqlite.Open(ctx, filepath.Join(t.TempDir(), "dogama.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
service := audit.New(db)
|
||||
if err := service.SetPolicy(ctx, audit.Policy{RetentionDays: -1}); err == nil {
|
||||
t.Fatal("negative retention accepted")
|
||||
}
|
||||
if _, err := service.Purge(ctx, time.Now().Add(time.Hour)); err == nil {
|
||||
t.Fatal("future purge accepted")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,425 @@
|
||||
// Package notification manages encrypted channels and bounded asynchronous delivery.
|
||||
package notification
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/aes"
|
||||
"crypto/cipher"
|
||||
"crypto/hmac"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"crypto/tls"
|
||||
"database/sql"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"net/smtp"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Channel struct {
|
||||
ID, Name, Type string
|
||||
Enabled bool
|
||||
Events []string
|
||||
Configured bool
|
||||
}
|
||||
type Input struct {
|
||||
Name, Type string
|
||||
Enabled bool
|
||||
Events []string
|
||||
Config map[string]string
|
||||
}
|
||||
type Event struct{ Type, Title, Message, InstanceName, OperationID string }
|
||||
type Service struct {
|
||||
db *sql.DB
|
||||
aead cipher.AEAD
|
||||
client *http.Client
|
||||
resolver *net.Resolver
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func New(db *sql.DB, key []byte) (*Service, error) {
|
||||
if len(key) != 32 {
|
||||
return nil, errors.New("notification encryption key must be exactly 32 bytes")
|
||||
}
|
||||
block, err := aes.NewCipher(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
aead, err := cipher.NewGCM(block)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s := &Service{db: db, aead: aead, resolver: net.DefaultResolver, now: time.Now}
|
||||
transport := http.DefaultTransport.(*http.Transport).Clone()
|
||||
transport.DialContext = s.dialSafe
|
||||
s.client = &http.Client{Transport: transport, Timeout: 10 * time.Second, CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
||||
if len(via) >= 3 {
|
||||
return errors.New("too many redirects")
|
||||
}
|
||||
return s.validateURL(req.Context(), req.URL)
|
||||
}}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Service) Upsert(ctx context.Context, id string, in Input) (Channel, error) {
|
||||
if strings.TrimSpace(in.Name) == "" || !validType(in.Type) {
|
||||
return Channel{}, errors.New("invalid notification channel")
|
||||
}
|
||||
if err := validateConfigShape(in.Type, in.Config); err != nil {
|
||||
return Channel{}, err
|
||||
}
|
||||
encrypted, err := s.seal(in.Config)
|
||||
if err != nil {
|
||||
return Channel{}, err
|
||||
}
|
||||
events, _ := json.Marshal(normalizeEvents(in.Events))
|
||||
now := s.now().UTC().Format(time.RFC3339Nano)
|
||||
if id == "" {
|
||||
id = randomID()
|
||||
}
|
||||
_, err = s.db.ExecContext(ctx, `INSERT INTO notification_channels(id,name,type,enabled,encrypted_config,event_filter_json,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET name=excluded.name,type=excluded.type,enabled=excluded.enabled,encrypted_config=excluded.encrypted_config,event_filter_json=excluded.event_filter_json,updated_at=excluded.updated_at`, id, strings.TrimSpace(in.Name), in.Type, in.Enabled, encrypted, string(events), now, now)
|
||||
if err != nil {
|
||||
return Channel{}, fmt.Errorf("save notification channel: %w", err)
|
||||
}
|
||||
return Channel{ID: id, Name: strings.TrimSpace(in.Name), Type: in.Type, Enabled: in.Enabled, Events: normalizeEvents(in.Events), Configured: true}, nil
|
||||
}
|
||||
func (s *Service) List(ctx context.Context) ([]Channel, error) {
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT id,name,type,enabled,event_filter_json,length(encrypted_config)>0 FROM notification_channels ORDER BY name COLLATE NOCASE`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Channel
|
||||
for rows.Next() {
|
||||
var c Channel
|
||||
var body string
|
||||
if err := rows.Scan(&c.ID, &c.Name, &c.Type, &c.Enabled, &body, &c.Configured); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_ = json.Unmarshal([]byte(body), &c.Events)
|
||||
out = append(out, c)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
func (s *Service) Delete(ctx context.Context, id string) error {
|
||||
_, err := s.db.ExecContext(ctx, `DELETE FROM notification_channels WHERE id=?`, id)
|
||||
return err
|
||||
}
|
||||
func (s *Service) Queue(ctx context.Context, event Event) error {
|
||||
if !validEvent(event.Type) || len(event.Message) > 1000 {
|
||||
return errors.New("invalid notification event")
|
||||
}
|
||||
payload, _ := json.Marshal(event)
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT id,event_filter_json FROM notification_channels WHERE enabled=1`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var channelIDs []string
|
||||
for rows.Next() {
|
||||
var id, filter string
|
||||
if err := rows.Scan(&id, &filter); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
var events []string
|
||||
_ = json.Unmarshal([]byte(filter), &events)
|
||||
if matches(events, event.Type) {
|
||||
channelIDs = append(channelIDs, id)
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
rows.Close()
|
||||
now := s.now().UTC().Format(time.RFC3339Nano)
|
||||
for _, id := range channelIDs {
|
||||
if _, err := s.db.ExecContext(ctx, `INSERT INTO notification_deliveries(id,channel_id,event_type,payload_redacted,next_attempt_at,created_at) VALUES(?,?,?,?,?,?)`, randomID(), id, event.Type, string(payload), now, now); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (s *Service) Test(ctx context.Context, id string) error {
|
||||
return s.queueForChannel(ctx, id, Event{Type: "notification.test", Title: "DoGaMa test notification", Message: "This is a test notification from DoGaMa."})
|
||||
}
|
||||
func (s *Service) queueForChannel(ctx context.Context, id string, event Event) error {
|
||||
var enabled bool
|
||||
if err := s.db.QueryRowContext(ctx, `SELECT enabled FROM notification_channels WHERE id=?`, id).Scan(&enabled); err != nil {
|
||||
return err
|
||||
}
|
||||
payload, _ := json.Marshal(event)
|
||||
now := s.now().UTC().Format(time.RFC3339Nano)
|
||||
_, err := s.db.ExecContext(ctx, `INSERT INTO notification_deliveries(id,channel_id,event_type,payload_redacted,next_attempt_at,created_at) VALUES(?,?,?,?,?,?)`, randomID(), id, event.Type, string(payload), now, now)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Service) RunDue(ctx context.Context) error {
|
||||
rows, err := s.db.QueryContext(ctx, `SELECT d.id,d.attempt,d.payload_redacted,c.type,c.encrypted_config FROM notification_deliveries d JOIN notification_channels c ON c.id=d.channel_id WHERE d.status IN ('queued','retrying') AND d.next_attempt_at<=? ORDER BY d.next_attempt_at LIMIT 20`, s.now().UTC().Format(time.RFC3339Nano))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
type job struct {
|
||||
id, typ, payload string
|
||||
attempt int
|
||||
encrypted []byte
|
||||
}
|
||||
var jobs []job
|
||||
for rows.Next() {
|
||||
var j job
|
||||
if err := rows.Scan(&j.id, &j.attempt, &j.payload, &j.typ, &j.encrypted); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
jobs = append(jobs, j)
|
||||
}
|
||||
rows.Close()
|
||||
for _, j := range jobs {
|
||||
config, e := s.open(j.encrypted)
|
||||
if e == nil {
|
||||
e = s.deliver(ctx, j.id, j.typ, config, []byte(j.payload))
|
||||
}
|
||||
attempt := j.attempt + 1
|
||||
if e == nil {
|
||||
if _, updateErr := s.db.ExecContext(ctx, `UPDATE notification_deliveries SET status='succeeded',attempt=?,completed_at=?,last_error_code='' WHERE id=?`, attempt, s.now().UTC().Format(time.RFC3339Nano), j.id); updateErr != nil {
|
||||
return updateErr
|
||||
}
|
||||
} else {
|
||||
status := "retrying"
|
||||
if attempt >= 5 {
|
||||
status = "failed"
|
||||
}
|
||||
delay := time.Duration(1<<min(attempt, 6)) * time.Minute
|
||||
_, _ = s.db.ExecContext(ctx, `UPDATE notification_deliveries SET status=?,attempt=?,next_attempt_at=?,last_error_code=? WHERE id=?`, status, attempt, s.now().UTC().Add(delay).Format(time.RFC3339Nano), errorCode(e), j.id)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (s *Service) deliver(ctx context.Context, id, typ string, c map[string]string, payload []byte) error {
|
||||
if typ == "email" {
|
||||
host := c["host"]
|
||||
port := c["port"]
|
||||
if port == "" {
|
||||
port = "587"
|
||||
}
|
||||
addr := net.JoinHostPort(host, port)
|
||||
var auth smtp.Auth
|
||||
if c["username"] != "" {
|
||||
auth = smtp.PlainAuth("", c["username"], c["password"], host)
|
||||
}
|
||||
msg := []byte("To: " + c["to"] + "\r\nSubject: DoGaMa notification\r\nContent-Type: application/json\r\n\r\n" + string(payload))
|
||||
return sendSMTP(ctx, addr, host, auth, c["from"], strings.Split(c["to"], ","), msg)
|
||||
}
|
||||
u, err := url.Parse(c["url"])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err = s.validateURL(ctx, u); err != nil {
|
||||
return err
|
||||
}
|
||||
body := payload
|
||||
if typ == "discord" {
|
||||
var event Event
|
||||
_ = json.Unmarshal(payload, &event)
|
||||
body, _ = json.Marshal(map[string]string{"content": event.Title + "\n" + event.Message})
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-DoGaMa-Event-ID", id)
|
||||
timestamp := strconv.FormatInt(s.now().Unix(), 10)
|
||||
req.Header.Set("X-DoGaMa-Timestamp", timestamp)
|
||||
if secret := c["signing_secret"]; secret != "" {
|
||||
mac := hmac.New(sha256.New, []byte(secret))
|
||||
_, _ = mac.Write([]byte(timestamp + "." + string(body)))
|
||||
req.Header.Set("X-DoGaMa-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil)))
|
||||
}
|
||||
resp, err := s.client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return fmt.Errorf("remote_status_%d", resp.StatusCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (s *Service) validateURL(ctx context.Context, u *url.URL) error {
|
||||
if u.Scheme != "https" || u.User != nil || u.Hostname() == "" {
|
||||
return errors.New("unsafe_destination")
|
||||
}
|
||||
ips, err := s.resolver.LookupNetIP(ctx, "ip", u.Hostname())
|
||||
if err != nil {
|
||||
return errors.New("destination_resolution_failed")
|
||||
}
|
||||
for _, ip := range ips {
|
||||
if unsafeIP(ip) {
|
||||
return errors.New("unsafe_destination")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) dialSafe(ctx context.Context, network, address string) (net.Conn, error) {
|
||||
host, port, err := net.SplitHostPort(address)
|
||||
if err != nil {
|
||||
return nil, errors.New("unsafe_destination")
|
||||
}
|
||||
ips, err := s.resolver.LookupNetIP(ctx, "ip", host)
|
||||
if err != nil || len(ips) == 0 {
|
||||
return nil, errors.New("destination_resolution_failed")
|
||||
}
|
||||
dialer := net.Dialer{Timeout: 10 * time.Second}
|
||||
for _, ip := range ips {
|
||||
if unsafeIP(ip) {
|
||||
return nil, errors.New("unsafe_destination")
|
||||
}
|
||||
connection, dialErr := dialer.DialContext(ctx, network, net.JoinHostPort(ip.String(), port))
|
||||
if dialErr == nil {
|
||||
return connection, nil
|
||||
}
|
||||
}
|
||||
return nil, errors.New("delivery_failed")
|
||||
}
|
||||
|
||||
func sendSMTP(ctx context.Context, address, host string, auth smtp.Auth, from string, recipients []string, message []byte) error {
|
||||
dialer := net.Dialer{Timeout: 10 * time.Second}
|
||||
connection, err := dialer.DialContext(ctx, "tcp", address)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
client, err := smtp.NewClient(connection, host)
|
||||
if err != nil {
|
||||
_ = connection.Close()
|
||||
return err
|
||||
}
|
||||
defer client.Close()
|
||||
if ok, _ := client.Extension("STARTTLS"); !ok {
|
||||
return errors.New("smtp_tls_required")
|
||||
}
|
||||
if err := client.StartTLS(&tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}); err != nil {
|
||||
return err
|
||||
}
|
||||
if auth != nil {
|
||||
if err := client.Auth(auth); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := client.Mail(from); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, recipient := range recipients {
|
||||
if err := client.Rcpt(strings.TrimSpace(recipient)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
w, err := client.Data()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = w.Write(message); err != nil {
|
||||
_ = w.Close()
|
||||
return err
|
||||
}
|
||||
if err = w.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return client.Quit()
|
||||
}
|
||||
func unsafeIP(ip netip.Addr) bool {
|
||||
return !ip.IsValid() || ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified()
|
||||
}
|
||||
func (s *Service) seal(config map[string]string) ([]byte, error) {
|
||||
body, _ := json.Marshal(config)
|
||||
nonce := make([]byte, s.aead.NonceSize())
|
||||
if _, err := rand.Read(nonce); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.aead.Seal(nonce, nonce, body, nil), nil
|
||||
}
|
||||
func (s *Service) open(body []byte) (map[string]string, error) {
|
||||
n := s.aead.NonceSize()
|
||||
if len(body) < n {
|
||||
return nil, errors.New("invalid encrypted channel")
|
||||
}
|
||||
plain, err := s.aead.Open(nil, body[:n], body[n:], nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out map[string]string
|
||||
err = json.Unmarshal(plain, &out)
|
||||
return out, err
|
||||
}
|
||||
func validType(v string) bool { return v == "email" || v == "webhook" || v == "discord" }
|
||||
func validEvent(v string) bool {
|
||||
return v == "notification.test" || strings.HasSuffix(v, ".failed") || strings.HasSuffix(v, ".completed") || strings.HasSuffix(v, ".required")
|
||||
}
|
||||
func validateConfigShape(typ string, c map[string]string) error {
|
||||
if typ == "email" {
|
||||
if c["host"] == "" || c["from"] == "" || c["to"] == "" {
|
||||
return errors.New("email host, from and to are required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
u, err := url.Parse(c["url"])
|
||||
if err != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil {
|
||||
return errors.New("an HTTPS webhook URL is required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func normalizeEvents(in []string) []string {
|
||||
seen := map[string]bool{}
|
||||
out := []string{}
|
||||
for _, v := range in {
|
||||
v = strings.TrimSpace(v)
|
||||
if validEvent(v) && !seen[v] {
|
||||
seen[v] = true
|
||||
out = append(out, v)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
func matches(filter []string, event string) bool {
|
||||
for _, v := range filter {
|
||||
if v == event || v == "*" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
func errorCode(err error) string {
|
||||
v := err.Error()
|
||||
if strings.HasPrefix(v, "remote_status_") {
|
||||
return v
|
||||
}
|
||||
switch v {
|
||||
case "unsafe_destination", "destination_resolution_failed":
|
||||
return v
|
||||
}
|
||||
return "delivery_failed"
|
||||
}
|
||||
func randomID() string {
|
||||
b := make([]byte, 18)
|
||||
_, _ = rand.Read(b)
|
||||
return base64.RawURLEncoding.EncodeToString(b)
|
||||
}
|
||||
func min(a, b int) int {
|
||||
if a < b {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package notification_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.zaynet.fr/DoGaMa/DoGaMa-serv/internal/notification"
|
||||
"git.zaynet.fr/DoGaMa/DoGaMa-serv/internal/persistence/sqlite"
|
||||
)
|
||||
|
||||
func TestChannelSecretsAreEncryptedAndWriteOnly(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
db, err := sqlite.Open(ctx, filepath.Join(t.TempDir(), "dogama.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
service, err := notification.New(db, bytes.Repeat([]byte{7}, 32))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
channel, err := service.Upsert(ctx, "", notification.Input{Name: "ops", Type: "webhook", Enabled: true, Events: []string{"backup.failed"}, Config: map[string]string{"url": "https://example.com/hook", "signing_secret": "highly-sensitive"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var encrypted []byte
|
||||
if err := db.QueryRowContext(ctx, `SELECT encrypted_config FROM notification_channels WHERE id=?`, channel.ID).Scan(&encrypted); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if strings.Contains(string(encrypted), "highly-sensitive") {
|
||||
t.Fatal("secret stored in plaintext")
|
||||
}
|
||||
channels, err := service.List(ctx)
|
||||
if err != nil || len(channels) != 1 || !channels[0].Configured {
|
||||
t.Fatalf("channels=%#v err=%v", channels, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeliveryBlocksPrivateWebhookAndRetriesWithRedactedError(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
db, err := sqlite.Open(ctx, filepath.Join(t.TempDir(), "dogama.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
service, _ := notification.New(db, bytes.Repeat([]byte{8}, 32))
|
||||
channel, err := service.Upsert(ctx, "", notification.Input{Name: "unsafe", Type: "webhook", Enabled: true, Events: []string{"backup.failed"}, Config: map[string]string{"url": "https://127.0.0.1/hook", "signing_secret": "never-leak"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := service.Queue(ctx, notification.Event{Type: "backup.failed", Title: "Backup failed", Message: "Operation failed", OperationID: "op-1"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := service.RunDue(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var status, code string
|
||||
var attempt int
|
||||
if err := db.QueryRowContext(ctx, `SELECT status,attempt,last_error_code FROM notification_deliveries WHERE channel_id=?`, channel.ID).Scan(&status, &attempt, &code); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if status != "retrying" || attempt != 1 || code != "unsafe_destination" || strings.Contains(code, "never-leak") {
|
||||
t.Fatalf("status=%s attempt=%d code=%q", status, attempt, code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRejectsMissingKeyAndInsecureURL(t *testing.T) {
|
||||
if _, err := notification.New(nil, []byte("short")); err == nil {
|
||||
t.Fatal("short key accepted")
|
||||
}
|
||||
ctx := context.Background()
|
||||
db, err := sqlite.Open(ctx, filepath.Join(t.TempDir(), "dogama.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
service, _ := notification.New(db, bytes.Repeat([]byte{9}, 32))
|
||||
if _, err := service.Upsert(ctx, "", notification.Input{Name: "bad", Type: "discord", Enabled: true, Config: map[string]string{"url": "http://example.com"}}); err == nil {
|
||||
t.Fatal("insecure URL accepted")
|
||||
}
|
||||
}
|
||||
@@ -24,10 +24,10 @@ func TestOpenAppliesMigrationsAndConfiguration(t *testing.T) {
|
||||
if err := db.QueryRow("SELECT COUNT(*) FROM schema_migrations").Scan(&count); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 8 {
|
||||
t.Fatalf("got %d migrations, want 8", count)
|
||||
if count != 9 {
|
||||
t.Fatalf("got %d migrations, want 9", count)
|
||||
}
|
||||
for _, table := range []string{"instance_memberships", "permission_overrides", "installation_requests", "backup_policies", "backups", "imports"} {
|
||||
for _, table := range []string{"instance_memberships", "permission_overrides", "installation_requests", "backup_policies", "backups", "imports", "notification_channels", "notification_deliveries", "audit_events"} {
|
||||
var found int
|
||||
if err := db.QueryRow("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?", table).Scan(&found); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -62,8 +62,8 @@ func TestOpenAppliesMigrationsAndConfiguration(t *testing.T) {
|
||||
if err := db.QueryRow("SELECT COUNT(*) FROM schema_migrations").Scan(&count); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 8 {
|
||||
t.Fatalf("reopened database has %d migrations, want 8", count)
|
||||
if count != 9 {
|
||||
t.Fatalf("reopened database has %d migrations, want 9", count)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
CREATE TABLE notification_channels (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
type TEXT NOT NULL CHECK (type IN ('email', 'webhook', 'discord')),
|
||||
enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)),
|
||||
encrypted_config BLOB NOT NULL,
|
||||
event_filter_json TEXT NOT NULL DEFAULT '[]',
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE notification_deliveries (
|
||||
id TEXT PRIMARY KEY,
|
||||
channel_id TEXT NOT NULL REFERENCES notification_channels(id) ON DELETE CASCADE,
|
||||
event_type TEXT NOT NULL,
|
||||
payload_redacted TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'retrying', 'succeeded', 'failed')),
|
||||
attempt INTEGER NOT NULL DEFAULT 0 CHECK (attempt >= 0),
|
||||
next_attempt_at TEXT NOT NULL,
|
||||
last_error_code TEXT NOT NULL DEFAULT '',
|
||||
created_at TEXT NOT NULL,
|
||||
completed_at TEXT
|
||||
);
|
||||
CREATE INDEX notification_deliveries_due_idx ON notification_deliveries(status, next_attempt_at);
|
||||
|
||||
CREATE TABLE audit_events (
|
||||
id TEXT PRIMARY KEY,
|
||||
occurred_at TEXT NOT NULL,
|
||||
actor_id TEXT REFERENCES users(id) ON DELETE SET NULL,
|
||||
actor_label TEXT NOT NULL,
|
||||
instance_id TEXT REFERENCES instances(id) ON DELETE SET NULL,
|
||||
action TEXT NOT NULL,
|
||||
outcome TEXT NOT NULL CHECK (outcome IN ('allowed', 'denied', 'failed')),
|
||||
summary_json TEXT NOT NULL DEFAULT '{}'
|
||||
);
|
||||
CREATE INDEX audit_events_time_idx ON audit_events(occurred_at DESC, id DESC);
|
||||
CREATE INDEX audit_events_filters_idx ON audit_events(actor_id, instance_id, action, outcome);
|
||||
|
||||
INSERT INTO system_settings(key, value_json, revision, updated_at)
|
||||
VALUES ('audit_policy', '{"retention_days":30,"maximum_count":10000}', 1, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
|
||||
Reference in New Issue
Block a user