410 lines
12 KiB
Go
410 lines
12 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/owen/vibedns/internal/models"
|
|
)
|
|
|
|
// --- Query log ----------------------------------------------------------
|
|
|
|
// InsertQueryLogs writes a batch of query log rows in one transaction. The
|
|
// query logger buffers rows in memory and calls this periodically so DNS
|
|
// resolution never waits on disk.
|
|
func (db *DB) InsertQueryLogs(ctx context.Context, entries []models.QueryLogEntry) error {
|
|
if len(entries) == 0 {
|
|
return nil
|
|
}
|
|
return db.InTx(ctx, func(tx *sql.Tx) error {
|
|
stmt, err := tx.PrepareContext(ctx, `
|
|
INSERT INTO query_logs (ts, client_ip, network_id, network_name, qname, qtype, rcode,
|
|
source, cache_hit, blocked, policy_id, policy_name, blacklist_id, blacklist_name,
|
|
matched_rule, protocol, duration_us, answer_count)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer stmt.Close()
|
|
for _, e := range entries {
|
|
_, err := stmt.ExecContext(ctx,
|
|
e.Timestamp.UnixMilli(), e.ClientIP, nullInt64(e.NetworkID), e.NetworkName,
|
|
e.QName, e.QType, e.Rcode, e.Source, boolInt(e.CacheHit), boolInt(e.Blocked),
|
|
nullInt64(e.PolicyID), e.PolicyName, nullInt64(e.BlacklistID), e.BlacklistName,
|
|
e.MatchedRule, e.Protocol, e.DurationUS, e.AnswerCount)
|
|
if err != nil {
|
|
return fmt.Errorf("write query log: %w", err)
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// QueryLogFilter narrows a query log search.
|
|
type QueryLogFilter struct {
|
|
Domain string
|
|
ClientIP string
|
|
QType string
|
|
Rcode string
|
|
Source string
|
|
Blocked string // "", "blocked", "allowed"
|
|
NetworkID int64
|
|
From time.Time
|
|
To time.Time
|
|
Limit int
|
|
Offset int
|
|
}
|
|
|
|
func (f QueryLogFilter) where() (string, []any) {
|
|
var conds []string
|
|
var args []any
|
|
if s := strings.TrimSpace(f.Domain); s != "" {
|
|
conds = append(conds, "qname LIKE ?")
|
|
args = append(args, "%"+strings.ToLower(s)+"%")
|
|
}
|
|
if s := strings.TrimSpace(f.ClientIP); s != "" {
|
|
conds = append(conds, "client_ip LIKE ?")
|
|
args = append(args, "%"+s+"%")
|
|
}
|
|
if s := strings.ToUpper(strings.TrimSpace(f.QType)); s != "" {
|
|
conds = append(conds, "qtype = ?")
|
|
args = append(args, s)
|
|
}
|
|
if s := strings.ToUpper(strings.TrimSpace(f.Rcode)); s != "" {
|
|
conds = append(conds, "rcode = ?")
|
|
args = append(args, s)
|
|
}
|
|
if s := strings.TrimSpace(f.Source); s != "" {
|
|
conds = append(conds, "source = ?")
|
|
args = append(args, s)
|
|
}
|
|
switch f.Blocked {
|
|
case "blocked":
|
|
conds = append(conds, "blocked = 1")
|
|
case "allowed":
|
|
conds = append(conds, "blocked = 0")
|
|
}
|
|
if f.NetworkID > 0 {
|
|
conds = append(conds, "network_id = ?")
|
|
args = append(args, f.NetworkID)
|
|
}
|
|
if !f.From.IsZero() {
|
|
conds = append(conds, "ts >= ?")
|
|
args = append(args, f.From.UnixMilli())
|
|
}
|
|
if !f.To.IsZero() {
|
|
conds = append(conds, "ts <= ?")
|
|
args = append(args, f.To.UnixMilli())
|
|
}
|
|
if len(conds) == 0 {
|
|
return "", nil
|
|
}
|
|
return " WHERE " + strings.Join(conds, " AND "), args
|
|
}
|
|
|
|
// QueryLogs returns matching rows newest first, plus the total match count.
|
|
func (db *DB) QueryLogs(ctx context.Context, f QueryLogFilter) ([]models.QueryLogEntry, int, error) {
|
|
whereSQL, args := f.where()
|
|
|
|
var total int
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM query_logs`+whereSQL, args...).Scan(&total); err != nil {
|
|
return nil, 0, fmt.Errorf("count query logs: %w", err)
|
|
}
|
|
|
|
q := `SELECT id, ts, client_ip, network_id, network_name, qname, qtype, rcode, source,
|
|
cache_hit, blocked, policy_id, policy_name, blacklist_id, blacklist_name, matched_rule,
|
|
protocol, duration_us, answer_count
|
|
FROM query_logs` + whereSQL + ` ORDER BY ts DESC, id DESC`
|
|
qargs := args
|
|
if f.Limit > 0 {
|
|
q += ` LIMIT ? OFFSET ?`
|
|
qargs = append(append([]any{}, args...), f.Limit, f.Offset)
|
|
}
|
|
|
|
rows, err := db.QueryContext(ctx, q, qargs...)
|
|
if err != nil {
|
|
return nil, 0, fmt.Errorf("read query logs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var out []models.QueryLogEntry
|
|
for rows.Next() {
|
|
var e models.QueryLogEntry
|
|
var ts int64
|
|
var netID, polID, blID sql.NullInt64
|
|
var cacheHit, blocked int
|
|
err := rows.Scan(&e.ID, &ts, &e.ClientIP, &netID, &e.NetworkName, &e.QName, &e.QType,
|
|
&e.Rcode, &e.Source, &cacheHit, &blocked, &polID, &e.PolicyName, &blID,
|
|
&e.BlacklistName, &e.MatchedRule, &e.Protocol, &e.DurationUS, &e.AnswerCount)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
e.Timestamp = time.UnixMilli(ts)
|
|
e.CacheHit = cacheHit != 0
|
|
e.Blocked = blocked != 0
|
|
if netID.Valid {
|
|
v := netID.Int64
|
|
e.NetworkID = &v
|
|
}
|
|
if polID.Valid {
|
|
v := polID.Int64
|
|
e.PolicyID = &v
|
|
}
|
|
if blID.Valid {
|
|
v := blID.Int64
|
|
e.BlacklistID = &v
|
|
}
|
|
out = append(out, e)
|
|
}
|
|
return out, total, rows.Err()
|
|
}
|
|
|
|
// PruneQueryLogs enforces the retention policy. Rows older than retentionDays
|
|
// are removed first, then the table is trimmed to maxRows newest entries.
|
|
// Either limit may be zero to disable it.
|
|
func (db *DB) PruneQueryLogs(ctx context.Context, retentionDays, maxRows int) (int64, error) {
|
|
var deleted int64
|
|
if retentionDays > 0 {
|
|
cutoff := time.Now().AddDate(0, 0, -retentionDays).UnixMilli()
|
|
res, err := db.ExecContext(ctx, `DELETE FROM query_logs WHERE ts < ?`, cutoff)
|
|
if err != nil {
|
|
return deleted, fmt.Errorf("prune query logs by age: %w", err)
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
deleted += n
|
|
}
|
|
if maxRows > 0 {
|
|
res, err := db.ExecContext(ctx, `
|
|
DELETE FROM query_logs WHERE id NOT IN (
|
|
SELECT id FROM query_logs ORDER BY ts DESC, id DESC LIMIT ?
|
|
)`, maxRows)
|
|
if err != nil {
|
|
return deleted, fmt.Errorf("prune query logs by count: %w", err)
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
deleted += n
|
|
}
|
|
return deleted, nil
|
|
}
|
|
|
|
// TruncateQueryLogs empties the query log table.
|
|
func (db *DB) TruncateQueryLogs(ctx context.Context) (int64, error) {
|
|
res, err := db.ExecContext(ctx, `DELETE FROM query_logs`)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("clear query logs: %w", err)
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
return n, nil
|
|
}
|
|
|
|
// NameCount is a domain/client aggregate used by the dashboard top-N lists.
|
|
type NameCount struct {
|
|
Name string `json:"name"`
|
|
Count int64 `json:"count"`
|
|
Extra string `json:"extra,omitempty"`
|
|
}
|
|
|
|
// TopQueried returns the most frequently queried names since `since`.
|
|
func (db *DB) TopQueried(ctx context.Context, since time.Time, limit int) ([]NameCount, error) {
|
|
return db.topBy(ctx, `SELECT qname, COUNT(*) c FROM query_logs WHERE ts >= ? GROUP BY qname ORDER BY c DESC LIMIT ?`,
|
|
since.UnixMilli(), limit)
|
|
}
|
|
|
|
// TopBlocked returns the most frequently blocked names since `since`.
|
|
func (db *DB) TopBlocked(ctx context.Context, since time.Time, limit int) ([]NameCount, error) {
|
|
return db.topBy(ctx, `SELECT qname, COUNT(*) c FROM query_logs WHERE blocked = 1 AND ts >= ? GROUP BY qname ORDER BY c DESC LIMIT ?`,
|
|
since.UnixMilli(), limit)
|
|
}
|
|
|
|
// TopClients returns the busiest clients since `since`.
|
|
func (db *DB) TopClients(ctx context.Context, since time.Time, limit int) ([]NameCount, error) {
|
|
rows, err := db.QueryContext(ctx, `
|
|
SELECT client_ip, COUNT(*) c, COALESCE(MAX(network_name), '')
|
|
FROM query_logs WHERE ts >= ? GROUP BY client_ip ORDER BY c DESC LIMIT ?`,
|
|
since.UnixMilli(), limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("top clients: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []NameCount
|
|
for rows.Next() {
|
|
var n NameCount
|
|
if err := rows.Scan(&n.Name, &n.Count, &n.Extra); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, n)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (db *DB) topBy(ctx context.Context, q string, args ...any) ([]NameCount, error) {
|
|
rows, err := db.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("aggregate query logs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []NameCount
|
|
for rows.Next() {
|
|
var n NameCount
|
|
if err := rows.Scan(&n.Name, &n.Count); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, n)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// TimeBucket is one point on the dashboard activity chart.
|
|
type TimeBucket struct {
|
|
Start time.Time `json:"start"`
|
|
Total int64 `json:"total"`
|
|
Blocked int64 `json:"blocked"`
|
|
Cached int64 `json:"cached"`
|
|
}
|
|
|
|
// ActivityBuckets groups query log rows into fixed-width time buckets covering
|
|
// the window [since, now].
|
|
func (db *DB) ActivityBuckets(ctx context.Context, since time.Time, bucket time.Duration, count int) ([]TimeBucket, error) {
|
|
if bucket <= 0 || count <= 0 {
|
|
return nil, nil
|
|
}
|
|
width := bucket.Milliseconds()
|
|
start := since.UnixMilli()
|
|
|
|
rows, err := db.QueryContext(ctx, `
|
|
SELECT (ts - ?) / ? AS b, COUNT(*), SUM(blocked), SUM(cache_hit)
|
|
FROM query_logs WHERE ts >= ?
|
|
GROUP BY b ORDER BY b`, start, width, start)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("bucket query logs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
buckets := make([]TimeBucket, count)
|
|
for i := range buckets {
|
|
buckets[i].Start = time.UnixMilli(start + int64(i)*width)
|
|
}
|
|
for rows.Next() {
|
|
var idx, total int64
|
|
var blocked, cached sql.NullInt64
|
|
if err := rows.Scan(&idx, &total, &blocked, &cached); err != nil {
|
|
return nil, err
|
|
}
|
|
if idx < 0 || idx >= int64(count) {
|
|
continue
|
|
}
|
|
buckets[idx].Total = total
|
|
buckets[idx].Blocked = blocked.Int64
|
|
buckets[idx].Cached = cached.Int64
|
|
}
|
|
return buckets, rows.Err()
|
|
}
|
|
|
|
// QueryLogCount returns the number of stored rows.
|
|
func (db *DB) QueryLogCount(ctx context.Context) (int64, error) {
|
|
var n int64
|
|
err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM query_logs`).Scan(&n)
|
|
return n, err
|
|
}
|
|
|
|
// --- Audit log ----------------------------------------------------------
|
|
|
|
// InsertAudit appends an administrative audit entry.
|
|
func (db *DB) InsertAudit(ctx context.Context, e models.AuditEntry) error {
|
|
if e.Timestamp.IsZero() {
|
|
e.Timestamp = time.Now()
|
|
}
|
|
_, err := db.ExecContext(ctx, `
|
|
INSERT INTO audit_logs (ts, actor, source, client_ip, action, object_type, object_id, object_name, details)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
e.Timestamp.UnixMilli(), e.Actor, e.Source, e.ClientIP, e.Action,
|
|
e.ObjectType, e.ObjectID, e.ObjectName, e.Details)
|
|
if err != nil {
|
|
return fmt.Errorf("write audit log: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// AuditFilter narrows an audit log search.
|
|
type AuditFilter struct {
|
|
Search string
|
|
ObjectType string
|
|
Source string
|
|
Limit int
|
|
Offset int
|
|
}
|
|
|
|
// AuditLogs returns audit entries newest first, plus the total match count.
|
|
func (db *DB) AuditLogs(ctx context.Context, f AuditFilter) ([]models.AuditEntry, int, error) {
|
|
var conds []string
|
|
var args []any
|
|
if s := strings.TrimSpace(f.Search); s != "" {
|
|
conds = append(conds, "(action LIKE ? OR object_name LIKE ? OR details LIKE ? OR actor LIKE ?)")
|
|
pat := "%" + s + "%"
|
|
args = append(args, pat, pat, pat, pat)
|
|
}
|
|
if f.ObjectType != "" {
|
|
conds = append(conds, "object_type = ?")
|
|
args = append(args, f.ObjectType)
|
|
}
|
|
if f.Source != "" {
|
|
conds = append(conds, "source = ?")
|
|
args = append(args, f.Source)
|
|
}
|
|
whereSQL := ""
|
|
if len(conds) > 0 {
|
|
whereSQL = " WHERE " + strings.Join(conds, " AND ")
|
|
}
|
|
|
|
var total int
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM audit_logs`+whereSQL, args...).Scan(&total); err != nil {
|
|
return nil, 0, fmt.Errorf("count audit logs: %w", err)
|
|
}
|
|
|
|
q := `SELECT id, ts, actor, source, client_ip, action, object_type, object_id, object_name, details
|
|
FROM audit_logs` + whereSQL + ` ORDER BY ts DESC, id DESC`
|
|
qargs := args
|
|
if f.Limit > 0 {
|
|
q += ` LIMIT ? OFFSET ?`
|
|
qargs = append(append([]any{}, args...), f.Limit, f.Offset)
|
|
}
|
|
|
|
rows, err := db.QueryContext(ctx, q, qargs...)
|
|
if err != nil {
|
|
return nil, 0, fmt.Errorf("read audit logs: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var out []models.AuditEntry
|
|
for rows.Next() {
|
|
var e models.AuditEntry
|
|
var ts int64
|
|
if err := rows.Scan(&e.ID, &ts, &e.Actor, &e.Source, &e.ClientIP, &e.Action,
|
|
&e.ObjectType, &e.ObjectID, &e.ObjectName, &e.Details); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
e.Timestamp = time.UnixMilli(ts)
|
|
out = append(out, e)
|
|
}
|
|
return out, total, rows.Err()
|
|
}
|
|
|
|
// PruneAuditLogs trims the audit log to the newest maxRows entries.
|
|
func (db *DB) PruneAuditLogs(ctx context.Context, maxRows int) (int64, error) {
|
|
if maxRows <= 0 {
|
|
return 0, nil
|
|
}
|
|
res, err := db.ExecContext(ctx, `
|
|
DELETE FROM audit_logs WHERE id NOT IN (
|
|
SELECT id FROM audit_logs ORDER BY ts DESC, id DESC LIMIT ?
|
|
)`, maxRows)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("prune audit logs: %w", err)
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
return n, nil
|
|
}
|