feat: audit ai config changes

This commit is contained in:
QiuSW
2026-07-02 11:42:39 +08:00
parent 18ac054dc5
commit c0020e8d29
16 changed files with 520 additions and 23 deletions
+136 -1
View File
@@ -1,7 +1,8 @@
from django import forms
from django.contrib import admin
from .models import AiModel, ModelAlias
from .audit import create_config_audit_log, snapshot_config
from .models import AiConfigAuditLog, AiModel, ModelAlias
from .security import AiKeyEncryptionError, encrypt_api_key
@@ -98,6 +99,53 @@ class AiModelAdmin(admin.ModelAdmin):
def api_key_status(self, obj):
return obj.api_key_masked or "not set"
def save_model(self, request, obj, form, change):
before = _snapshot_existing(obj) if change else None
super().save_model(request, obj, form, change)
create_config_audit_log(
actor=request.user,
action=(
AiConfigAuditLog.Action.UPDATE
if change
else AiConfigAuditLog.Action.CREATE
),
instance=obj,
before=before,
after=snapshot_config(obj),
)
def delete_model(self, request, obj):
before = snapshot_config(obj)
target_id = obj.pk
target_repr = str(obj)
super().delete_model(request, obj)
create_config_audit_log(
actor=request.user,
action=AiConfigAuditLog.Action.DELETE,
instance=obj,
before=before,
after=None,
target_id=target_id,
target_repr=target_repr,
)
def delete_queryset(self, request, queryset):
snapshots = [
(obj, obj.pk, str(obj), snapshot_config(obj))
for obj in queryset
]
super().delete_queryset(request, queryset)
for obj, target_id, target_repr, before in snapshots:
create_config_audit_log(
actor=request.user,
action=AiConfigAuditLog.Action.DELETE,
instance=obj,
before=before,
after=None,
target_id=target_id,
target_repr=target_repr,
)
@admin.register(ModelAlias)
class ModelAliasAdmin(admin.ModelAdmin):
@@ -113,3 +161,90 @@ class ModelAliasAdmin(admin.ModelAdmin):
search_fields = ("alias", "ai_model__name", "ai_model__model")
autocomplete_fields = ("ai_model",)
readonly_fields = ("created_at", "updated_at")
def save_model(self, request, obj, form, change):
before = _snapshot_existing(obj) if change else None
super().save_model(request, obj, form, change)
create_config_audit_log(
actor=request.user,
action=(
AiConfigAuditLog.Action.UPDATE
if change
else AiConfigAuditLog.Action.CREATE
),
instance=obj,
before=before,
after=snapshot_config(obj),
)
def delete_model(self, request, obj):
before = snapshot_config(obj)
target_id = obj.pk
target_repr = str(obj)
super().delete_model(request, obj)
create_config_audit_log(
actor=request.user,
action=AiConfigAuditLog.Action.DELETE,
instance=obj,
before=before,
after=None,
target_id=target_id,
target_repr=target_repr,
)
def delete_queryset(self, request, queryset):
snapshots = [
(obj, obj.pk, str(obj), snapshot_config(obj))
for obj in queryset
]
super().delete_queryset(request, queryset)
for obj, target_id, target_repr, before in snapshots:
create_config_audit_log(
actor=request.user,
action=AiConfigAuditLog.Action.DELETE,
instance=obj,
before=before,
after=None,
target_id=target_id,
target_repr=target_repr,
)
@admin.register(AiConfigAuditLog)
class AiConfigAuditLogAdmin(admin.ModelAdmin):
list_display = (
"created_at",
"actor",
"action",
"target_type",
"target_id",
"target_repr",
"changed_fields_display",
)
list_filter = ("action", "target_type", "created_at")
search_fields = (
"target_repr",
"actor__username",
"actor__email",
)
ordering = ("-created_at", "-id")
def get_readonly_fields(self, request, obj=None):
return tuple(field.name for field in self.model._meta.fields)
def has_add_permission(self, request):
return False
def has_change_permission(self, request, obj=None):
return False
def has_delete_permission(self, request, obj=None):
return False
@admin.display(description="changed fields")
def changed_fields_display(self, obj):
return ", ".join(obj.changed_fields)
def _snapshot_existing(obj):
return snapshot_config(obj.__class__.objects.get(pk=obj.pk))
+111
View File
@@ -0,0 +1,111 @@
from __future__ import annotations
from typing import Any
from django.db import models
from .models import AiConfigAuditLog, AiModel, ModelAlias
AI_MODEL_AUDIT_FIELDS = (
"name",
"url",
"model",
"api_type",
"api_key_encrypted",
"capabilities",
"timeout_seconds",
"connect_timeout_seconds",
"extra_body",
"is_active",
)
MODEL_ALIAS_AUDIT_FIELDS = (
"alias",
"operation_type",
"ai_model_id",
"is_default",
"is_active",
)
FIELD_ALIASES = {"api_key_encrypted": "api_key", "ai_model_id": "ai_model"}
def create_config_audit_log(
*,
actor: Any,
action: str,
instance: AiModel | ModelAlias,
before: dict[str, Any] | None,
after: dict[str, Any] | None,
target_id: int | None = None,
target_repr: str | None = None,
) -> AiConfigAuditLog | None:
"""Create a sanitized audit log for AI config changes."""
changed_fields, changes = _diff_snapshots(before, after)
if action == AiConfigAuditLog.Action.UPDATE and not changes:
return None
actor_value = actor if getattr(actor, "is_authenticated", False) else None
return AiConfigAuditLog.objects.create(
actor=actor_value,
action=action,
target_type=_target_type(instance),
target_id=target_id if target_id is not None else instance.pk,
target_repr=(target_repr or str(instance))[:255],
changed_fields=changed_fields,
changes=changes,
)
def snapshot_config(instance: AiModel | ModelAlias) -> dict[str, Any]:
fields = _audit_fields(instance)
return {field: _field_value(instance, field) for field in fields}
def _diff_snapshots(
before: dict[str, Any] | None,
after: dict[str, Any] | None,
) -> tuple[list[str], dict[str, dict[str, Any]]]:
fields = tuple((after or before or {}).keys())
changed_fields = []
changes = {}
for field in fields:
old_value = None if before is None else before.get(field)
new_value = None if after is None else after.get(field)
if old_value == new_value:
continue
public_field = FIELD_ALIASES.get(field, field)
changed_fields.append(public_field)
changes[public_field] = {
"old": _sanitize_value(field, old_value),
"new": _sanitize_value(field, new_value),
}
return changed_fields, changes
def _target_type(instance: AiModel | ModelAlias) -> str:
if isinstance(instance, AiModel):
return AiConfigAuditLog.TargetType.AI_MODEL
if isinstance(instance, ModelAlias):
return AiConfigAuditLog.TargetType.MODEL_ALIAS
raise TypeError(f"unsupported audit target: {type(instance)!r}")
def _audit_fields(instance: AiModel | ModelAlias) -> tuple[str, ...]:
if isinstance(instance, AiModel):
return AI_MODEL_AUDIT_FIELDS
if isinstance(instance, ModelAlias):
return MODEL_ALIAS_AUDIT_FIELDS
raise TypeError(f"unsupported audit target: {type(instance)!r}")
def _field_value(instance: models.Model, field: str) -> Any:
value = getattr(instance, field)
if isinstance(value, (list, dict, str, int, bool)) or value is None:
return value
return str(value)
def _sanitize_value(field: str, value: Any) -> Any:
if field == "api_key_encrypted":
return "set" if value else "empty"
return value
@@ -0,0 +1,35 @@
# Generated by Django 5.2.15 on 2026-07-02 03:14
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('ai', '0001_initial'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='AiConfigAuditLog',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('action', models.CharField(choices=[('create', 'Create'), ('update', 'Update'), ('delete', 'Delete')], max_length=16)),
('target_type', models.CharField(choices=[('ai_model', 'AI model'), ('model_alias', 'Model alias')], max_length=32)),
('target_id', models.PositiveBigIntegerField(blank=True, null=True)),
('target_repr', models.CharField(max_length=255)),
('changed_fields', models.JSONField(blank=True, default=list)),
('changes', models.JSONField(blank=True, default=dict)),
('created_at', models.DateTimeField(auto_now_add=True)),
('actor', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to=settings.AUTH_USER_MODEL)),
],
options={
'db_table': 'ai_config_audit_log',
'ordering': ('-created_at', '-id'),
'indexes': [models.Index(fields=['target_type', 'target_id'], name='ai_config_a_target__3e1809_idx'), models.Index(fields=['action', 'created_at'], name='ai_config_a_action_924966_idx'), models.Index(fields=['actor', 'created_at'], name='ai_config_a_actor_i_a11e67_idx')],
},
),
]
+39
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import models
@@ -133,3 +134,41 @@ class ModelAlias(models.Model):
def save(self, *args, **kwargs) -> None:
self.full_clean()
super().save(*args, **kwargs)
class AiConfigAuditLog(models.Model):
class TargetType(models.TextChoices):
AI_MODEL = "ai_model", "AI model"
MODEL_ALIAS = "model_alias", "Model alias"
class Action(models.TextChoices):
CREATE = "create", "Create"
UPDATE = "update", "Update"
DELETE = "delete", "Delete"
actor = models.ForeignKey(
settings.AUTH_USER_MODEL,
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="+",
)
action = models.CharField(max_length=16, choices=Action.choices)
target_type = models.CharField(max_length=32, choices=TargetType.choices)
target_id = models.PositiveBigIntegerField(null=True, blank=True)
target_repr = models.CharField(max_length=255)
changed_fields = models.JSONField(default=list, blank=True)
changes = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
db_table = "ai_config_audit_log"
ordering = ("-created_at", "-id")
indexes = [
models.Index(fields=("target_type", "target_id")),
models.Index(fields=("action", "created_at")),
models.Index(fields=("actor", "created_at")),
]
def __str__(self) -> str:
return f"{self.created_at:%Y-%m-%d %H:%M:%S} {self.action} {self.target_repr}"
+137 -2
View File
@@ -2,12 +2,13 @@ import base64
from cryptography.fernet import Fernet
from django.contrib.admin.sites import AdminSite
from django.contrib.auth import get_user_model
from django.test import RequestFactory, SimpleTestCase, TestCase, override_settings
from apps.ai.admin import AiModelAdmin
from apps.ai.admin import AiConfigAuditLogAdmin, AiModelAdmin, ModelAliasAdmin
from apps.ai.aliases import AliasNotFoundError, ModelCapabilityError, resolve_alias
from apps.ai.importers import import_ai_models_config
from apps.ai.models import AiModel, ModelAlias
from apps.ai.models import AiConfigAuditLog, AiModel, ModelAlias
from apps.ai.providers import AiCapabilityError, ResolvedModel, get_provider, resolve_api_type
from apps.ai.providers.openai_compatible import ChatCompletionsProvider, ImagesEditsProvider
@@ -425,3 +426,137 @@ class AiModelAdminTests(TestCase):
self.assertFalse(form.is_valid())
self.assertIn("AI_KEY_ENCRYPTION_KEY is not configured", str(form.errors))
@override_settings(AI_KEY_ENCRYPTION_KEY=TEST_ENCRYPTION_KEY)
class AiConfigAuditAdminTests(TestCase):
def setUp(self):
self.site = AdminSite()
self.request = RequestFactory().post("/admin/")
self.request.user = get_user_model().objects.create_superuser(
username="auditor",
email="auditor@example.com",
password="password",
)
def create_ai_model(
self,
*,
name="GPT-5.5 text",
model="gpt-5.5",
capabilities=None,
api_type=AiModel.ApiType.CHAT,
url="https://api.vectorengine.ai/v1",
api_key="sk-test-secret",
):
ai_model = AiModel(
name=name,
url=url,
model=model,
api_type=api_type,
capabilities=capabilities or ["text"],
)
ai_model.set_api_key(api_key)
ai_model.save()
return ai_model
def test_aimodel_admin_create_writes_sanitized_audit_log(self):
ai_model = AiModel(
name="GPT-5.5 text",
url="https://api.vectorengine.ai/v1",
model="gpt-5.5",
api_type=AiModel.ApiType.CHAT,
capabilities=["text"],
)
ai_model.set_api_key("sk-created-secret")
AiModelAdmin(AiModel, self.site).save_model(
self.request,
ai_model,
form=None,
change=False,
)
log = AiConfigAuditLog.objects.get()
self.assertEqual(log.actor, self.request.user)
self.assertEqual(log.action, AiConfigAuditLog.Action.CREATE)
self.assertEqual(log.target_type, AiConfigAuditLog.TargetType.AI_MODEL)
self.assertIn("api_key", log.changed_fields)
self.assertEqual(log.changes["api_key"], {"old": "empty", "new": "set"})
self.assertNotIn("sk-created-secret", str(log.changes))
self.assertNotIn("fernet:", str(log.changes))
def test_aimodel_admin_update_logs_field_and_key_changes(self):
ai_model = self.create_ai_model()
ai_model.url = "https://api.vectorengine.ai/v2"
ai_model.set_api_key("sk-new-secret")
AiModelAdmin(AiModel, self.site).save_model(
self.request,
ai_model,
form=None,
change=True,
)
log = AiConfigAuditLog.objects.get()
self.assertEqual(log.action, AiConfigAuditLog.Action.UPDATE)
self.assertEqual(set(log.changed_fields), {"url", "api_key"})
self.assertEqual(
log.changes["url"],
{
"old": "https://api.vectorengine.ai/v1",
"new": "https://api.vectorengine.ai/v2",
},
)
self.assertEqual(log.changes["api_key"], {"old": "set", "new": "set"})
self.assertNotIn("sk-new-secret", str(log.changes))
def test_model_alias_admin_update_logs_mapping_change(self):
text_model = self.create_ai_model(name="Text model", model="gpt-5.5")
image_model = self.create_ai_model(
name="Image model",
model="gpt-image-2",
capabilities=["image"],
api_type=AiModel.ApiType.IMAGES_EDITS,
url="https://api.vectorengine.ai/v1/images/edits",
)
alias = ModelAlias.objects.create(
operation_type=ModelAlias.OperationType.IMAGE,
alias="image-standard",
ai_model=text_model,
)
alias.ai_model = image_model
ModelAliasAdmin(ModelAlias, self.site).save_model(
self.request,
alias,
form=None,
change=True,
)
log = AiConfigAuditLog.objects.get()
self.assertEqual(log.target_type, AiConfigAuditLog.TargetType.MODEL_ALIAS)
self.assertEqual(log.changed_fields, ["ai_model"])
self.assertEqual(
log.changes["ai_model"],
{"old": text_model.id, "new": image_model.id},
)
def test_aimodel_admin_delete_writes_audit_log(self):
ai_model = self.create_ai_model()
target_id = ai_model.id
AiModelAdmin(AiModel, self.site).delete_model(self.request, ai_model)
log = AiConfigAuditLog.objects.get()
self.assertEqual(log.action, AiConfigAuditLog.Action.DELETE)
self.assertEqual(log.target_id, target_id)
self.assertIn("api_key", log.changed_fields)
self.assertEqual(log.changes["api_key"], {"old": "set", "new": "empty"})
def test_audit_log_admin_is_read_only(self):
model_admin = AiConfigAuditLogAdmin(AiConfigAuditLog, self.site)
self.assertFalse(model_admin.has_add_permission(self.request))
self.assertFalse(model_admin.has_change_permission(self.request))
self.assertFalse(model_admin.has_delete_permission(self.request))
self.assertIn("changes", model_admin.get_readonly_fields(self.request))