feat(api): list prompt evidence and notes for moderators

- Add moderator-only GET /api/v2/operations/moderation/prompts (OperationsPromptEvents), a newest-first page of events with their notes, filtered by account, proxied account, address, reporting worker and capture time, and paged by before_id; the page reports the retention windows in force in its retention field.
- Match the ipaddr filter on the address pseudonym, falling back to the address only on events without one, so an event stays findable after its address is removed; a blank value filters nothing and a value that is not one IP subject is refused with 400 BadRequest, like the OperationsIP endpoints.
- Expose the pseudonym as the opaque ip_subject_key, filterable by exact value so a moderator can pivot on it after the address is gone, and a non-address origin as origin_text; an event_id filter serves the alert link; the listing and PromptModerationEvent API model are explicit field lists, so a column never reaches moderators by accident.
- Add POST /prompts/<id>/notes (OperationsPromptNotes, add_prompt_note), locking the event so the note is ordered against a concurrent anonymization; a note on an anonymized event is stored, and a note exempts its event from further retention steps.
- Limit each moderator resource to 120 reads and 30 writes a minute per API key and method (get_request_api_key_per_method), so notes on every event share one budget, and expose the rate limit headers to cross-origin clients.
- Add the InvalidOperationsLimit, InvalidOperationsCursor, InvalidModerationTimeRange, InvalidModerationNote, InvalidModerationAddressFilter, InvalidModerationEventID and ModerationEventNotFound return codes.
This commit is contained in:
tazlin authored and GitHub committed 2026-10-01 17:52:36 -04:00
1 parent e1b739e2d0
commit d9a730beba
10 files changed
+1137 -18

No files matched your search

+7
View File
@@ -174,6 +174,12 @@ The errors returned by the AI horde are always in this json format
| InvalidControlType | The annotation form requires a known control_type in its payload |
| InvalidExtraSourceImages | This request type does not accept extra source images |
| InvalidExtraTexts | This request type does not accept extra texts |
| InvalidModerationAddressFilter | The moderation evidence `ipaddr` or `ip_subject_key` filter is not a valid address, network or pseudonym |
| InvalidModerationEventID | The moderation evidence `event_id` filter is not a positive integer |
| InvalidModerationNote | A moderation note is 1 to 2000 characters and may not contain NUL bytes |
| InvalidModerationTimeRange | The moderation evidence time bounds are not a valid range |
| InvalidOperationsCursor | The pagination cursor for a moderator operations listing is not positive |
| InvalidOperationsLimit | The page size for a moderator operations listing is out of range |
| InvalidPromptEncoding | The prompt contains characters that cannot be encoded (unpaired UTF-16 surrogates) |
| InvalidRemixModel | Image remix is not available for this model baseline |
| InvalidTransparencyCN | Transparent images cannot be generated in ControlNet workflows |
@@ -182,6 +188,7 @@ The errors returned by the AI horde are always in this json format
| MessagesOnlyOwnWorkers | Messages can only be sent to your own workers |
| MissingExtraTexts | This workflow requires the extra texts to be provided |
| MissingFullSamplerOrder | A custom sampler order must list every sampler |
| ModerationEventNotFound | No moderation event exists with the given ID |
| MoreThanMinExtraSourceImage | Only trusted users and patrons can send more than one extra source image |
| NonServiceForbidden | Only service accounts can set a proxy passkey |
| NotUserOrMod | Only the user themselves or a moderator can delete a user |
+10
View File
@@ -57,6 +57,16 @@ def get_request_api_key():
return f"{apikey}@{request.method}@{request.path}"
def get_request_api_key_per_method():
"""Key a limit by the hashed API key and method alone.
A decorated limit is already scoped to its route's endpoint, so leaving the concrete path out makes a route with a
path parameter, such as one event's notes, share one budget across every value of the parameter.
"""
apikey = hash_api_key(request.headers.get("apikey", "0000000000"))
return f"{apikey}@{request.method}"
def get_request_limit_per_apikey():
apikey = request.headers.get("apikey", "0000000000")
if apikey == "0000000000" or dynamic_ip_whitelist.is_ip_whitelisted(get_remoteaddr()):
+119
View File
@@ -2,8 +2,12 @@
#
# SPDX-License-Identifier: AGPL-3.0-or-later
from datetime import UTC, datetime
from typing import Any, override
from flask_restx import fields, reqparse
from horde.classes.base.prompt_moderation import MAX_NOTE_CHARACTERS, PromptModerationOutcome, PromptModerationReason
from horde.enums import WarningMessage
from horde.exceptions import KNOWN_RC
from horde.vars import horde_noun, horde_title
@@ -42,6 +46,23 @@ def derive_submitted_request_model(api, name, input_model, overrides=None):
return api.model(name, fields_by_name)
class UTCDateTime(fields.DateTime):
"""ISO 8601 timestamp that always carries an explicit ``+00:00`` offset.
Timestamps are stored as naive UTC. flask-restx formats a naive value without an offset, which clients then read
as their local time, so a naive value is marked as UTC before formatting.
"""
def __init__(self, **kwargs: Any) -> None:
super().__init__(dt_format="iso8601", **kwargs)
@override
def format_iso8601(self, dt: datetime) -> str:
if dt.tzinfo is None:
dt = dt.replace(tzinfo=UTC)
return dt.astimezone(UTC).isoformat()
class Parsers:
def __init__(self):
# A Basic parser which only expects a Client-Agent
@@ -1774,6 +1795,104 @@ class Models:
),
},
)
self.response_model_prompt_moderation_note = api.model(
"PromptModerationNote",
{
"id": fields.Integer(description="Note identifier."),
"author_id": fields.Integer(description="Moderator who wrote the note."),
"note": fields.String(description=f"Note text, at most {MAX_NOTE_CHARACTERS} characters."),
"created": UTCDateTime(description="UTC time the note was written."),
},
)
self.input_model_prompt_moderation_note = api.model(
"PromptModerationNoteInput",
{
"note": fields.String(required=True, description=f"Note text, 1 to {MAX_NOTE_CHARACTERS} characters."),
},
)
self.response_model_prompt_moderation_event = api.model(
"PromptModerationEvent",
{
"id": fields.Integer(description="Evidence identifier, also the pagination cursor."),
"created": UTCDateTime(description="UTC capture time in ISO 8601 with a +00:00 offset."),
"user_id": fields.Integer(description="Submitting account; null once the event is anonymized."),
"request_id": fields.String(description="Waiting request ID; null for a rejected submission."),
"worker_id": fields.String(
description="Worker that reported the job, for worker reports; it identifies the reporter, not the subject.",
),
"proxied_account": fields.String(description="Service-account subject, when one was supplied."),
"ipaddr": fields.String(
description="Submitting address; an IPv6 address is recorded as its /64 network.",
),
"ip_subject_key": fields.String(
description="Address pseudonym, equal for events from the same IPv4 address or IPv6 /64; it "
"outlives ipaddr and is null once the event is anonymized. Group or filter events by it; it is not "
"an address.",
),
"origin_text": fields.String(
description="The origin as a trusted proxy reported it when it was not an IP address; null "
"otherwise or once the address window has passed.",
),
"models": fields.List(fields.String, description="Model names the request asked for."),
"reason": fields.String(
description="Why the evidence was retained.",
enum=[reason.value for reason in PromptModerationReason],
),
"outcome": fields.String(
description="What happened to the request.",
enum=[outcome.value for outcome in PromptModerationOutcome],
),
"submitted_prompt": fields.String(description="Original submission; null when provenance is unknown."),
"moderation_prompt": fields.String(description="Moderation input after style expansion, when known."),
"effective_prompt": fields.String(description="Prompt a worker received; null for a rejection."),
"text_truncated": fields.Boolean(description="A prompt stage exceeded the evidence length limit."),
"text_redacted": fields.Boolean(
description="Retention ran its text step on the event, past the text window or at anonymization, "
"whether or not there was prompt text to remove.",
),
"ipaddr_redacted": fields.Boolean(
description="Retention ran its address step on the event, past the address window or at "
"anonymization, whether or not there was an address to remove.",
),
"anonymized": fields.Boolean(
description="Retention removed the account, proxied account, address, address pseudonym, request "
"and job identifiers and prompt stages; notes are kept.",
),
"notes": fields.List(
fields.Nested(self.response_model_prompt_moderation_note),
description="Moderator notes, oldest first.",
),
},
)
self.response_model_prompt_moderation_retention = api.model(
"PromptModerationRetention",
{
"text_days": fields.Integer(
description="Days until prompt stages are removed from an event not under moderation action; null "
"keeps them until anonymization.",
),
"ipaddr_days": fields.Integer(
description="Days until the address is removed from an event not under moderation action; null "
"keeps it until anonymization.",
),
"ceiling_days": fields.Integer(
description="Days until an event not under moderation action is anonymized; the reason, outcome, "
"models, capture time, reporting worker and notes are kept. An event with a note, or of a flagged "
"or suspicious account, is kept whole.",
),
},
)
self.response_model_prompt_moderation_events = api.model(
"PromptModerationEvents",
{
"events": fields.List(fields.Nested(self.response_model_prompt_moderation_event)),
"next_cursor": fields.Integer(description="Pass as before_id to read the next page; null on the last page."),
"retention": fields.Nested(
self.response_model_prompt_moderation_retention,
description="Retention windows in force, counted from capture; they apply only to events not under moderation action.",
),
},
)
self.input_model_add_ip_timeout = api.model(
"AddTimeoutIPInput",
{
+3
View File
@@ -5,6 +5,7 @@
import horde.apis.v2.base as base
import horde.apis.v2.kobold as kobold
import horde.apis.v2.kobold_styles as kobold_styles
import horde.apis.v2.moderation as moderation
import horde.apis.v2.stable as stable
import horde.apis.v2.stable_styles as stable_styles
import horde.apis.v2.styles as styles
@@ -57,6 +58,8 @@ api.add_resource(base.TeamSingle, "/teams/<string:team_id>")
api.add_resource(base.OperationsIP, "/operations/ipaddr")
api.add_resource(base.OperationsIPSingle, "/operations/ipaddr/<string:ipaddr>")
api.add_resource(base.OperationsBlockWorkerIP, "/operations/block_worker_ipaddr/<string:worker_id>")
api.add_resource(moderation.OperationsPromptEvents, "/operations/moderation/prompts")
api.add_resource(moderation.OperationsPromptNotes, "/operations/moderation/prompts/<int:event_id>/notes")
api.add_resource(stable.Interrogate, "/interrogate/async")
api.add_resource(stable.InterrogationStatus, "/interrogate/status/<string:id>")
api.add_resource(stable.InterrogatePop, "/interrogate/pop")
+1
View File
@@ -156,6 +156,7 @@ handle_request_not_found = api.errorhandler(e.RequestNotFound)(e.handle_bad_requ
handle_worker_not_found = api.errorhandler(e.WorkerNotFound)(e.handle_bad_requests)
handle_team_not_found = api.errorhandler(e.TeamNotFound)(e.handle_bad_requests)
handle_thing_not_found = api.errorhandler(e.ThingNotFound)(e.handle_bad_requests)
handle_moderation_event_not_found = api.errorhandler(e.ModerationEventNotFound)(e.handle_bad_requests)
handle_user_not_found = api.errorhandler(e.UserNotFound)(e.handle_bad_requests)
handle_duplicate_gen = api.errorhandler(e.DuplicateGen)(e.handle_bad_requests)
handle_aborted_gen = api.errorhandler(e.AbortedGen)(e.handle_bad_requests)
+274
View File
@@ -0,0 +1,274 @@
# SPDX-FileCopyrightText: 2026 Tazlin
#
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Expose moderator-only review of retained prompt evidence.
Every resource here requires a moderator ``apikey`` and answers with
``Cache-Control: private, no-store``: the payloads carry account identity,
IP addresses, and retained prompt text.
"""
from __future__ import annotations
import re
from datetime import UTC, datetime
from typing import Any
from flask_restx import Resource, reqparse
import horde.apis.limiter_api as lim
from horde import exceptions as e
from horde.apis.v2.base import api, check_for_mod, models
from horde.classes.base.prompt_moderation import MAX_NOTE_CHARACTERS
from horde.countermeasures import CounterMeasures
from horde.database import prompt_moderation as evidence_db
from horde.flask import db
from horde.limiter import limiter
PRIVATE_HEADERS: dict[str, str] = {"Cache-Control": "private, no-store"}
"""The caching headers of every response here.
The payloads carry account identity, addresses and prompt text, so no shared or browser cache may keep them.
"""
MAX_EVIDENCE_PAGE: int = 100
"""The most events one evidence page returns.
An event carries up to three prompt stages of ``MAX_EVIDENCE_CHARACTERS`` each, so 100 events bound a page near five
million characters of prompt text.
"""
IP_SUBJECT_KEY_PATTERN: re.Pattern[str] = re.compile(r"[0-9a-f]{64}")
"""The form of an address pseudonym, the hex HMAC-SHA256 digest ``CounterMeasures.ip_subject_key`` returns."""
def _moderator_limits() -> list[Any]:
"""Return the rate limits of a moderator resource: 120 reads and 30 writes a minute per API key and resource.
Moderators share office addresses and VPN exits, so the count follows the key. Each limit is scoped to its
resource's endpoint, so all of one resource's paths, such as the notes of every event, share one budget. A request
without a key counts against the anonymous key's bucket, and the missing header refuses it anyway.
"""
return [
limiter.limit("120/minute", key_func=lim.get_request_api_key_per_method, methods=["GET"]),
limiter.limit(
"30/minute",
key_func=lim.get_request_api_key_per_method,
methods=[
"POST",
"PUT",
"PATCH",
"DELETE",
],
),
]
def _add_moderator_credentials(parser: reqparse.RequestParser) -> None:
"""Add the moderator key and client agent headers every resource here requires."""
parser.add_argument("apikey", type=str, required=True, help="A mod API key.", location="headers")
parser.add_argument(
"Client-Agent",
default="unknown:0:unknown",
type=str,
required=False,
help="The client name and version.",
location="headers",
)
def _parse_utc_bound(value: str | None, field: str) -> datetime | None:
"""Convert an ISO 8601 bound with an explicit timezone into a naive UTC datetime.
Raises:
e.BadRequest: The value is not ISO 8601 or carries no timezone.
"""
if value is None:
return None
try:
parsed = datetime.fromisoformat(value)
except ValueError as err:
raise e.BadRequest(f"{field} must be an ISO 8601 timestamp", rc="InvalidModerationTimeRange") from err
if parsed.tzinfo is None:
raise e.BadRequest(f"{field} must carry an explicit timezone", rc="InvalidModerationTimeRange")
return parsed.astimezone(UTC).replace(tzinfo=None)
def _assert_page_bounds(limit: int, before_id: int | None, maximum: int) -> None:
"""Reject a page size or descending cursor outside the documented range.
Raises:
e.BadRequest: The page size or the cursor is out of range.
"""
if not 1 <= limit <= maximum:
raise e.BadRequest(f"limit must be between 1 and {maximum}", rc="InvalidOperationsLimit")
if before_id is not None and before_id < 1:
raise e.BadRequest("before_id must be positive", rc="InvalidOperationsCursor")
def _parse_ipaddr_filter(value: str | None) -> str | None:
"""Parse the ``ipaddr`` filter into an IP subject; a missing or blank value restricts nothing.
Raises:
e.BadRequest: The value is neither an address nor an IPv6 network of /64 or narrower.
"""
if value is None or not value.strip():
return None
try:
return CounterMeasures.parse_ip_subject(value)
except ValueError as err:
raise e.BadRequest(
"ipaddr must be an IP address or an IPv6 network of /64 or narrower",
rc="InvalidModerationAddressFilter",
) from err
def _assert_ip_subject_key(value: str | None) -> str | None:
"""Validate the ``ip_subject_key`` filter.
Raises:
e.BadRequest: The value is not 64 lowercase hexadecimal characters.
"""
if value is not None and IP_SUBJECT_KEY_PATTERN.fullmatch(value) is None:
raise e.BadRequest("ip_subject_key must be 64 lowercase hexadecimal characters", rc="InvalidModerationAddressFilter")
return value
def _assert_note(note: str) -> None:
if not note.strip() or len(note) > MAX_NOTE_CHARACTERS or "\x00" in note:
raise e.BadRequest(
f"note must be 1 to {MAX_NOTE_CHARACTERS} characters and contain no NUL characters",
rc="InvalidModerationNote",
)
class OperationsPromptEvents(Resource):
"""Return actionable evidence; ordinary filter replacements are not events."""
decorators = _moderator_limits()
get_parser = reqparse.RequestParser()
_add_moderator_credentials(get_parser)
get_parser.add_argument("limit", type=int, default=50, help="Rows per page (1-100).", location="args")
get_parser.add_argument("before_id", type=int, required=False, help="Exclusive descending cursor.", location="args")
get_parser.add_argument(
"event_id",
type=int,
required=False,
help="Restrict to one event; the event_id of a moderator alert link.",
location="args",
)
get_parser.add_argument("user_id", type=int, required=False, help="Restrict to one account.", location="args")
get_parser.add_argument(
"proxied_account",
type=str,
required=False,
help="Restrict to one proxied account.",
location="args",
)
get_parser.add_argument(
"ipaddr",
type=str,
required=False,
help="Restrict to one IP subject, including events whose address retention removed; an IPv6 address selects its /64.",
location="args",
)
get_parser.add_argument(
"ip_subject_key",
type=str,
required=False,
help="Restrict to one address pseudonym, the ip_subject_key of an event.",
location="args",
)
get_parser.add_argument("worker_id", type=str, required=False, help="Restrict to one reporting worker.", location="args")
get_parser.add_argument("since", type=str, required=False, help="Inclusive ISO 8601 lower bound.", location="args")
get_parser.add_argument("until", type=str, required=False, help="Exclusive ISO 8601 upper bound.", location="args")
@api.expect(get_parser)
@api.marshal_with(
models.response_model_prompt_moderation_events,
code=200,
description="Retained prompt moderation evidence",
)
@api.response(400, "Validation Error", models.response_model_error)
@api.response(401, "Invalid API Key", models.response_model_error)
@api.response(403, "Access Denied", models.response_model_error)
def get(self) -> tuple[dict[str, Any], int, dict[str, str]]:
"""Return a bounded evidence page filtered by subject, time, or ID range."""
self.args = self.get_parser.parse_args()
check_for_mod(self.args.apikey, "GET OperationsPromptEvents")
_assert_page_bounds(self.args.limit, self.args.before_id, MAX_EVIDENCE_PAGE)
since = _parse_utc_bound(self.args.since, "since")
until = _parse_utc_bound(self.args.until, "until")
if since and until and since >= until:
raise e.BadRequest("since must be earlier than until", rc="InvalidModerationTimeRange")
if self.args.event_id is not None and self.args.event_id < 1:
raise e.BadRequest("event_id must be positive", rc="InvalidModerationEventID")
events = evidence_db.get_prompt_events(
limit=self.args.limit,
before_id=self.args.before_id,
event_id=self.args.event_id,
user_id=self.args.user_id,
proxied_account=self.args.proxied_account,
ip_subject=_parse_ipaddr_filter(self.args.ipaddr),
ip_subject_key=_assert_ip_subject_key(self.args.ip_subject_key),
worker_id=self.args.worker_id,
since=since,
until=until,
)
policy = evidence_db.MODERATION_RETENTION_POLICY
events["retention"] = {
"text_days": policy.text_days,
"ipaddr_days": policy.ipaddr_days,
"ceiling_days": policy.ceiling_days,
}
# The page is fully materialized; release the pooled connection before marshalling.
db.session.remove()
return events, 200, PRIVATE_HEADERS
class OperationsPromptNotes(Resource):
"""Attach moderator notes to an event without modifying the evidence."""
decorators = _moderator_limits()
post_parser = reqparse.RequestParser()
_add_moderator_credentials(post_parser)
# Not required here: _assert_note refuses a missing note too, so every invalid note reports InvalidModerationNote.
post_parser.add_argument(
"note",
type=str,
required=False,
help=f"Note text, 1 to {MAX_NOTE_CHARACTERS} characters.",
location="json",
)
@api.expect(post_parser, models.input_model_prompt_moderation_note)
@api.marshal_with(models.response_model_prompt_moderation_note, code=201, description="The stored note")
@api.response(400, "Validation Error", models.response_model_error)
@api.response(401, "Invalid API Key", models.response_model_error)
@api.response(403, "Access Denied", models.response_model_error)
@api.response(404, "Moderation Event Not Found", models.response_model_error)
def post(self, event_id: int) -> tuple[evidence_db.NoteRecord, int, dict[str, str]]:
"""Add a note to an event.
Args:
event_id: Evidence identifier returned by the event listing.
Returns:
The stored note.
Raises:
e.BadRequest: The note is missing, empty, too long, or carries a NUL byte.
e.ModerationEventNotFound: The event does not exist.
"""
self.args = self.post_parser.parse_args()
moderator = check_for_mod(self.args.apikey, "POST OperationsPromptNotes")
note = self.args.note or ""
_assert_note(note)
author_id = moderator.id
# add_prompt_note owns its own transaction; release this request's first.
db.session.remove()
stored = evidence_db.add_prompt_note(event_id=event_id, author_id=author_id, note=note)
if stored is None:
raise e.ModerationEventNotFound(event_id)
return stored, 201, PRIVATE_HEADERS
+160 -17
View File
@@ -2,7 +2,7 @@
#
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Capture actionable prompt evidence and decay it on schedule.
"""Capture actionable prompt evidence, list it for moderators, and decay it on schedule.
Evidence writes use a separate transaction, with no foreign-key dependency on
the submitting transaction. A rejected submission rollback cannot erase them.
@@ -24,17 +24,19 @@ retention pass and the privacy document read it when they run. ``EVENT_COLUMN_RE
from __future__ import annotations
import os
from collections import defaultdict
from collections.abc import Collection, Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum, auto
from types import MappingProxyType
from typing import Any
from typing import Any, TypedDict
from sqlalchemy import ColumnElement, Select, and_, delete, exists, func, or_, select, update
from sqlalchemy.dialects.postgresql import insert as postgres_insert
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from horde.classes.base.prompt_moderation import (
MAX_ORIGIN_TEXT_CHARACTERS,
@@ -51,39 +53,29 @@ from horde.logger import logger
TEXT_RETENTION_ENV: str = "HORDE_MODERATION_TEXT_RETENTION_DAYS"
"""The environment variable for text retention."""
IPADDR_RETENTION_ENV: str = "HORDE_MODERATION_IPADDR_RETENTION_DAYS"
"""The environment variable for IP address retention."""
EVIDENCE_CEILING_ENV: str = "HORDE_MODERATION_EVIDENCE_CEILING_DAYS"
"""The environment variable for the evidence ceiling."""
DEFAULT_TEXT_RETENTION_DAYS: int | None = None
"""The default text retention window in days, or None to keep text until the ceiling.
Text is kept until the ceiling by default so that evidence no moderator has reviewed yet stays reviewable for the whole
evidence period. An operator can set a shorter window.
"""
DEFAULT_IPADDR_RETENTION_DAYS: int = 30
"""The default IP address retention window in days.
The raw address serves the review of recent activity. After the window the address pseudonym still links the event
to the same IP subject, so the address itself is not kept.
"""
DEFAULT_EVIDENCE_CEILING_DAYS: int = 365
"""The default evidence ceiling in days, the longest ``MAX_EVIDENCE_CEILING_DAYS`` permits.
The ceiling anonymizes only events not under moderation action.
"""
MAX_EVIDENCE_CEILING_DAYS: int = 365
"""The maximum evidence ceiling in days.
@@ -91,11 +83,9 @@ Identifying data on an event that no moderation action has touched is kept for a
moderation action are exempt from the ceiling, so it bounds only records that serve no enforcement purpose.
"""
NO_SEPARATE_WINDOW: str = "none"
"""Indicate that there is no separate retention window for this type of data."""
MAX_EVIDENCE_CHARACTERS: int = 16000
"""The most characters of each prompt stage an event keeps.
@@ -103,7 +93,6 @@ It bounds the size of an event row and of a listing page, which returns up to 10
longer stage is clipped and the event marked ``text_truncated``.
"""
MAX_PROXIED_ACCOUNT_CHARACTERS: int = 1000
"""The most characters of a proxied account an event keeps.
@@ -112,7 +101,6 @@ request never reaches that column, so this keeps every value an accepted request
refused one writes.
"""
CLEANUP_BATCH_SIZE: int = 1000
"""The most rows each retention step changes or deletes in one pass.
@@ -278,7 +266,6 @@ a column of an event under moderation action, and notes are separate rows no ste
a column without a fate, so a new column cannot be retained or cleared by accident.
"""
REMOVAL_FLAGS: Mapping[RetentionFate, str] = MappingProxyType(
{
RetentionFate.TEXT: "text_redacted",
@@ -384,6 +371,19 @@ class PromptEvidence:
"""The model names the request asked for."""
class NoteRecord(TypedDict):
"""Represent a moderator note as the listing and the note endpoint return it."""
id: int
"""The ID of the note record."""
author_id: int
"""The ID of the author of the note."""
note: str
"""The content of the note."""
created: datetime
"""The timestamp when the note was created."""
def subjectless_origin_text(origin: str | None, ip_subject: str | None) -> str | None:
"""Return the origin text evidence keeps for an origin that is not blank and has no IP subject.
@@ -456,6 +456,149 @@ def record_prompt_evidence(evidence: PromptEvidence) -> int | None:
return None
def _ip_subject_criterion(ip_subject: str) -> ColumnElement[bool]:
"""Return the filter selecting the events of one IP subject.
An event with an address pseudonym matches by it alone. An event without one, captured while the deployment had no
usable secret, matches by its address. Each branch has its own partial index.
"""
unkeyed_match = and_(PromptModerationEvent.ip_subject_key.is_(None), PromptModerationEvent.ipaddr == ip_subject)
subject_key = CounterMeasures.ip_subject_key(ip_subject)
if subject_key is None:
return unkeyed_match
return or_(PromptModerationEvent.ip_subject_key == subject_key, unkeyed_match)
def get_prompt_events(
*,
limit: int,
before_id: int | None = None,
event_id: int | None = None,
user_id: int | None = None,
proxied_account: str | None = None,
ip_subject: str | None = None,
ip_subject_key: str | None = None,
worker_id: str | None = None,
since: datetime | None = None,
until: datetime | None = None,
) -> dict[str, Any]:
"""Return a bounded, newest-first page of evidence with its notes.
Args:
limit: Maximum number of events in the page.
before_id: Exclusive descending cursor; only events with a lower ID are returned.
event_id: Restrict to one event.
user_id: Restrict to one account.
proxied_account: Restrict to one proxied account.
ip_subject: Restrict to one IP subject (``CounterMeasures.parse_ip_subject``). It matches by address
pseudonym, so an event stays selected after its address is removed.
ip_subject_key: Restrict to one address pseudonym, matched exactly.
worker_id: Restrict to one reporting worker.
since: Inclusive naive UTC lower bound on the capture time.
until: Exclusive naive UTC upper bound on the capture time.
Returns:
``events``, each with its notes oldest first, and ``next_cursor``, the ``before_id`` of the next page or None.
"""
statement = select(PromptModerationEvent)
if before_id is not None:
statement = statement.where(PromptModerationEvent.id < before_id)
if event_id is not None:
statement = statement.where(PromptModerationEvent.id == event_id)
if user_id is not None:
statement = statement.where(PromptModerationEvent.user_id == user_id)
if proxied_account is not None:
statement = statement.where(PromptModerationEvent.proxied_account == proxied_account)
if ip_subject is not None:
statement = statement.where(_ip_subject_criterion(ip_subject))
if ip_subject_key is not None:
statement = statement.where(PromptModerationEvent.ip_subject_key == ip_subject_key)
if worker_id is not None:
statement = statement.where(PromptModerationEvent.worker_id == worker_id)
if since is not None:
statement = statement.where(PromptModerationEvent.created >= since)
if until is not None:
statement = statement.where(PromptModerationEvent.created < until)
statement = statement.order_by(PromptModerationEvent.id.desc()).limit(limit + 1)
rows = list(db.session.execute(statement).scalars())
notes: dict[int, list[NoteRecord]] = defaultdict(list)
if rows[:limit]:
for note in db.session.execute(
select(PromptModerationNote)
.where(PromptModerationNote.event_id.in_([evidence.id for evidence in rows[:limit]]))
.order_by(PromptModerationNote.id),
).scalars():
notes[note.event_id].append(_note_record(note))
events = []
for evidence in rows[:limit]:
# Deliberate allowlist: never serialize ORM __dict__ or arbitrary relationships.
events.append(
{
"id": evidence.id,
"created": evidence.created,
"user_id": evidence.user_id,
"request_id": evidence.request_id,
"worker_id": evidence.worker_id,
"proxied_account": evidence.proxied_account,
"ipaddr": evidence.ipaddr,
"ip_subject_key": evidence.ip_subject_key,
"origin_text": evidence.origin_text,
"models": evidence.models,
"reason": evidence.reason,
"outcome": evidence.outcome,
"submitted_prompt": evidence.submitted_prompt,
"moderation_prompt": evidence.moderation_prompt,
"effective_prompt": evidence.effective_prompt,
"text_truncated": evidence.text_truncated,
"text_redacted": evidence.text_redacted,
"ipaddr_redacted": evidence.ipaddr_redacted,
"anonymized": evidence.anonymized,
"notes": notes.get(evidence.id, []),
}
)
return {
"events": events,
"next_cursor": events[-1]["id"] if len(rows) > limit else None,
}
def _note_record(note: PromptModerationNote) -> NoteRecord:
return {
"id": note.id,
"author_id": note.author_id,
"note": note.note,
"created": note.created,
}
def add_prompt_note(*, event_id: int, author_id: int, note: str) -> NoteRecord | None:
"""Attach a note to an event.
An anonymized event takes notes too, and they are stored. Postgres checks the note's foreign key with a key-share
lock on the event row, held until commit, which orders the note against ``_anonymize_batch``.
Args:
event_id: Event the note is about.
author_id: Moderator writing the note.
note: Validated note text.
Returns:
The stored note, or None when the event does not exist.
"""
# Own this transaction explicitly; unrelated ORM mutations must not be committed here.
# Own this transaction explicitly; unrelated ORM mutations must not be committed here. Events are never deleted,
# so an event that exists here still exists when the note is inserted.
with Session(db.engine, expire_on_commit=False) as session, session.begin():
event = session.execute(
select(PromptModerationEvent.id).where(PromptModerationEvent.id == event_id),
).scalar_one_or_none()
if event is None:
return None
record = PromptModerationNote(event_id=event_id, author_id=author_id, note=note)
session.add(record)
return _note_record(record)
def _account_restricted(user_id: ColumnElement[int]) -> ColumnElement[bool]:
"""Return whether the account ``user_id`` identifies is flagged, or suspicious as ``User.is_suspicious`` judges it.
+14
View File
@@ -173,6 +173,13 @@ KNOWN_RC = [
"SamplerKnobOutOfRange",
"SamplerSolverTypeUnsupported",
"SamplerSchedulerMismatch",
"InvalidOperationsLimit",
"InvalidOperationsCursor",
"InvalidModerationTimeRange",
"InvalidModerationAddressFilter",
"InvalidModerationEventID",
"InvalidModerationNote",
"ModerationEventNotFound",
"SchedulerBaselineMismatch",
"FlowShiftInapplicable",
"FlowShiftOutOfRange",
@@ -533,6 +540,13 @@ class TeamNotFound(wze.NotFound):
self.rc = rc
class ModerationEventNotFound(wze.NotFound):
def __init__(self, event_id, rc="ModerationEventNotFound"):
self.specific = f"Moderation event with ID '{event_id}' not found."
self.log = None
self.rc = rc
class ThingNotFound(wze.NotFound):
def __init__(self, thing_type, thing_id, message=None, rc="ThingNotFound"):
if message is not None:
+3
View File
@@ -189,6 +189,9 @@ def create_app(config: dict[str, object] | None = None) -> Flask:
"Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, apikey, "
"Client-Agent, X-Fields, X-Forwarded-For, Proxied-For, Proxy-Authorization"
)
# A browser hides every other response header from cross-origin scripts, and a client waiting out a 429 needs
# the rate limit headers flask-limiter sets.
response.headers["Access-Control-Expose-Headers"] = "Retry-After, X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset"
response.headers["Horde-Node"] = f"{socket.gethostname()}:{args.port}:{HORDE_VERSION}"
if response.content_type == "application/json":
+546 -1
View File
@@ -2,7 +2,7 @@
#
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Exercise prompt provenance, rollback-safe evidence, tiered retention and account wipes.
"""Exercise prompt provenance, rollback-safe evidence, tiered retention, account wipes, and the moderator evidence endpoints.
Retention skips events under moderation action, so that exemption is exercised with each retention step.
@@ -23,20 +23,40 @@ from sqlalchemy import select
from horde.classes.base.prompt_moderation import (
MAX_ORIGIN_TEXT_CHARACTERS,
PromptModerationEvent,
PromptModerationNote,
PromptModerationReason,
)
from horde.classes.base.user import UserProblemJobs
from horde.countermeasures import CounterMeasures
from horde.database.prompt_moderation import (
EVENT_COLUMN_RETENTION,
PromptEvidence,
RetentionFate,
RetentionPassResult,
RetentionPolicy,
apply_evidence_retention,
get_prompt_events,
record_prompt_evidence,
)
from tests.fixture_types import ApiUser, MakeApiUser
from tests.integration.test_request_parameters import IMAGE_REQUEST
EVENTS_URL = "/api/v2/operations/moderation/prompts"
"""The moderator evidence listing."""
UNLISTED_EVENT_COLUMNS = frozenset({"job_id"})
"""Event columns the listing omits; the job ID is internal deduplication state."""
LISTING_ONLY_EVENT_KEYS = frozenset({"notes"})
"""Listed event keys that are not event columns."""
LISTED_EVENT_COLUMN_NAMES = {"ip_subject_key": "ip_subject"}
"""Event columns the listing exposes under another name; the pseudonym is an opaque token, not a key to reuse."""
TEST_IP_SUBJECT_SECRET = b"integration-test deployment secret"
"""The private secret the tests key address pseudonyms with, since the test environment sets none."""
@@ -236,6 +256,51 @@ def test_failed_replacement_is_recorded(client, app, make_api_user, settle_kudos
assert event.models == IMAGE_REQUEST["models"]
def test_evidence_survives_rollback_and_notes_are_separate(client, app, api_key, make_api_user) -> None:
from horde.classes.base.user import User
from horde.flask import db
submitter = make_api_user()
with app.app_context():
account = db.session.get(User, submitter.id)
old_name = account.username
account.username = "uncommitted-change"
db.session.flush()
event_id = record_prompt_evidence(
PromptEvidence(
user_id=submitter.id,
reason=PromptModerationReason.FILTER_REJECTION,
submitted_prompt="original",
moderation_prompt="styled original",
effective_prompt=None,
)
)
assert event_id is not None
db.session.rollback()
assert db.session.get(User, submitter.id).username == old_name
assert db.session.get(PromptModerationEvent, event_id).submitted_prompt == "original"
notes_url = f"{EVENTS_URL}/{event_id}/notes"
assert client.get(EVENTS_URL).status_code == 400
assert client.get(EVENTS_URL, headers={"apikey": submitter.api_key}).status_code == 403
assert client.post(notes_url, json={"note": "x"}, headers={"apikey": submitter.api_key}).status_code == 403
headers = {"apikey": api_key}
note = client.post(notes_url, json={"note": "synthetic false positive"}, headers=headers)
assert note.status_code == 201, note.get_json()
assert note.headers["Cache-Control"] == "private, no-store"
assert set(note.get_json()) == {"id", "author_id", "note", "created"}
page = client.get(EVENTS_URL, query_string={"user_id": submitter.id}, headers=headers)
assert page.status_code == 200, page.get_json()
assert page.headers["Cache-Control"] == "private, no-store"
(event,) = page.get_json()["events"]
assert event["notes"] == [note.get_json()]
with app.app_context():
assert db.session.get(PromptModerationEvent, event_id).submitted_prompt == "original"
stored = db.session.get(PromptModerationNote, note.get_json()["id"])
assert stored.event_id == event_id
assert stored.author_id == note.get_json()["author_id"]
def test_worker_problem_jobs_preserve_evidence_without_duplicate_rows(app, make_api_user) -> None:
from horde.classes.base.user import User
from horde.classes.kobold.worker import TextWorker
@@ -496,6 +561,71 @@ def test_an_origin_that_is_not_an_address_is_kept_as_text_without_address_or_pse
assert (problem_job.ipaddr, problem_job.origin_text) == (None, clipped)
def test_event_listing_exposes_documented_fields_with_utc_offsets(client, app, api_key, make_api_user, monkeypatch) -> None:
"""Each listed event carries exactly the documented field allowlist, and stored naive UTC times carry a zero offset.
The page also reports the retention windows in force, with null for a window kept until the ceiling.
"""
from horde.database import prompt_moderation
monkeypatch.setattr(prompt_moderation, "MODERATION_RETENTION_POLICY", _policy(text=None, ipaddr=12, ceiling=90))
submitter = make_api_user()
(event_id,) = _seed(app, user_id=submitter.id)
headers = {"apikey": api_key}
note = client.post(f"{EVENTS_URL}/{event_id}/notes", json={"note": "checked"}, headers=headers)
assert note.status_code == 201, note.get_json()
page = client.get(EVENTS_URL, query_string={"user_id": submitter.id}, headers=headers)
assert page.status_code == 200, page.get_json()
assert set(page.get_json()) == {"events", "next_cursor", "retention"}
assert page.get_json()["retention"] == {"text_days": None, "ipaddr_days": 12, "ceiling_days": 90}
(event,) = page.get_json()["events"]
assert event["id"] == event_id
assert (event["text_redacted"], event["ipaddr_redacted"], event["anonymized"]) == (False, False, False)
assert set(event) == {
"id",
"created",
"user_id",
"request_id",
"worker_id",
"proxied_account",
"ipaddr",
"ip_subject_key",
"origin_text",
"models",
"reason",
"outcome",
"submitted_prompt",
"moderation_prompt",
"effective_prompt",
"text_truncated",
"text_redacted",
"ipaddr_redacted",
"anonymized",
"notes",
}
for value in (event["created"], event["notes"][0]["created"]):
assert value.endswith(("+00:00", "Z")), value
assert datetime.fromisoformat(value).utcoffset() == timedelta(0), value
def test_event_listing_and_api_model_carry_every_column_except_the_unlisted(app, make_api_user) -> None:
"""A new event column fails this until it is listed and documented, or declared unlisted.
The listing and the API model are explicit field lists, so a column never reaches moderators by accident, and one
meant for them is never silently missing.
"""
from horde.apis.v2.base import models
listed_columns = set(PromptModerationEvent.__table__.columns.keys()) - UNLISTED_EVENT_COLUMNS
expected = listed_columns | LISTING_ONLY_EVENT_KEYS
submitter = make_api_user()
_seed(app, user_id=submitter.id)
with app.app_context():
(listed,) = get_prompt_events(limit=1, user_id=submitter.id)["events"]
assert set(listed) == expected
assert set(models.response_model_prompt_moderation_event) == expected
def test_worker_csam_submission_records_event(client, app, make_api_user, settle_kudos, monkeypatch) -> None:
"""An image job submitted with csam censorship metadata records one worker_csam event for that job and worker."""
from horde.classes.stable import processing_generation, waiting_prompt
@@ -691,6 +821,310 @@ def test_text_and_address_are_removed_after_their_windows(app, make_api_user) ->
assert not event.anonymized
@pytest.mark.usefixtures("_no_prior_evidence")
def test_anonymization_removes_identity_and_keeps_the_record_and_its_notes(client, app, api_key, make_api_user) -> None:
"""Past the ceiling an event keeps its reason, outcome, models, capture time, reporting worker and notes.
A note written on the anonymized event is stored, and a later pass keeps it.
"""
from horde.flask import db
submitter = make_api_user()
worker_id = str(uuid4())
expired = _record_aged(
app,
user_id=submitter.id,
age_days=31,
reason=PromptModerationReason.WORKER_CSAM,
effective_prompt="effective",
request_id=str(uuid4()),
job_id=str(uuid4()),
worker_id=worker_id,
)
kept = _record_aged(app, user_id=submitter.id, age_days=1)
with app.app_context():
# Capture stores an address or an origin text, never both; setting both covers every removable column.
db.session.get(PromptModerationEvent, expired).origin_text = "client behind proxy"
db.session.commit()
headers = {"apikey": api_key}
kept_note = client.post(f"{EVENTS_URL}/{kept}/notes", json={"note": "unrelated"}, headers=headers)
assert kept_note.status_code == 201, kept_note.get_json()
before = _stored(app, expired)
removable_fates = (RetentionFate.TEXT, RetentionFate.IPADDR, RetentionFate.IDENTITY)
for column, fate in EVENT_COLUMN_RETENTION.items():
if fate in removable_fates:
assert getattr(before, column) is not None, column
policy = _policy(text=20, ipaddr=10, ceiling=30)
with app.app_context():
assert apply_evidence_retention(policy) == _pass_result(anonymized=1)
assert apply_evidence_retention(policy).total == 0
assert db.session.get(PromptModerationNote, kept_note.get_json()["id"]) is not None
after = _stored(app, expired)
for identifying in (
"user_id",
"proxied_account",
"ipaddr",
"request_id",
"job_id",
"submitted_prompt",
"moderation_prompt",
"effective_prompt",
):
assert getattr(after, identifying) is None, identifying
assert after.text_redacted and after.ipaddr_redacted and after.anonymized
for column, fate in EVENT_COLUMN_RETENTION.items():
if fate in removable_fates:
assert getattr(after, column) is None, column
elif fate == RetentionFate.KEPT:
assert getattr(after, column) == getattr(before, column), column
else:
assert getattr(after, column) is True, column
assert (after.reason, after.outcome, after.models, after.created, after.text_truncated, after.worker_id) == (
"worker_csam",
"censored",
["model a"],
before.created,
before.text_truncated,
worker_id,
)
assert _stored(app, kept).user_id == submitter.id
(listed,) = client.get(EVENTS_URL, query_string={"worker_id": worker_id}, headers=headers).get_json()["events"]
assert listed["id"] == expired
assert listed["user_id"] is None
assert listed["ip_subject_key"] is None
assert (listed["text_redacted"], listed["ipaddr_redacted"], listed["anonymized"]) == (True, True, True)
assert listed["notes"] == []
@pytest.mark.usefixtures("_no_prior_evidence")
def test_a_note_on_an_anonymized_event_is_stored(client, app, api_key, make_api_user) -> None:
"""An anonymized event takes a note, and later passes keep the note and change nothing else."""
from horde.flask import db
submitter = make_api_user()
worker_id = str(uuid4())
expired = _record_aged(app, user_id=submitter.id, age_days=31, worker_id=worker_id)
policy = _policy(text=20, ipaddr=10, ceiling=30)
with app.app_context():
assert apply_evidence_retention(policy) == _pass_result(anonymized=1)
headers = {"apikey": api_key}
late_note = client.post(f"{EVENTS_URL}/{expired}/notes", json={"note": "after anonymization"}, headers=headers)
assert late_note.status_code == 201, late_note.get_json()
with app.app_context():
assert apply_evidence_retention(policy) == _pass_result()
assert db.session.get(PromptModerationNote, late_note.get_json()["id"]) is not None
(listed,) = client.get(EVENTS_URL, query_string={"worker_id": worker_id}, headers=headers).get_json()["events"]
assert listed["anonymized"]
assert listed["notes"] == [late_note.get_json()]
def _flag(app, user_id: int) -> None:
"""Flag the account, as a moderator does."""
from horde.classes.base.user import User
from horde.flask import db
with app.app_context():
user = db.session.get(User, user_id)
user.set_flagged(True)
db.session.expire_all()
assert db.session.get(User, user_id).flagged
def _add_suspicions(app, user_id: int, count: int) -> None:
"""Give the account ``count`` distinct suspicions."""
from horde.classes.base.user import UserSuspicions
from horde.flask import db
with app.app_context():
for suspicion_id in range(count):
db.session.add(UserSuspicions(user_id=user_id, suspicion_id=suspicion_id))
db.session.commit()
def _make_suspicious(app, user_id: int) -> None:
"""Give the account the suspicions ``User.is_suspicious`` needs to report it suspicious."""
from horde.classes.base.user import User
from horde.flask import db
_add_suspicions(app, user_id, User.SUSPICION_THRESHOLD)
with app.app_context():
assert db.session.get(User, user_id).is_suspicious()
@pytest.mark.usefixtures("_no_prior_evidence")
@pytest.mark.parametrize("action", ["note", "flagged", "suspicious"])
def test_events_under_moderation_action_are_skipped_by_every_retention_step(client, app, api_key, make_api_user, action: str) -> None:
"""An event with a note, or of a flagged or suspicious account, keeps every value past every window and the ceiling.
An unactioned event of the same age is anonymized in the same pass, so the pass did run.
"""
submitter, bystander = make_api_user(), make_api_user()
actioned = _record_aged(app, user_id=submitter.id, age_days=400)
unactioned = _record_aged(app, user_id=bystander.id, age_days=400)
if action == "note":
note = client.post(f"{EVENTS_URL}/{actioned}/notes", json={"note": "under review"}, headers={"apikey": api_key})
assert note.status_code == 201, note.get_json()
elif action == "flagged":
_flag(app, submitter.id)
else:
_make_suspicious(app, submitter.id)
before = _stored(app, actioned)
policy = _policy(text=10, ipaddr=10, ceiling=30)
with app.app_context():
assert apply_evidence_retention(policy) == _pass_result(anonymized=1)
assert apply_evidence_retention(policy) == _pass_result()
after = _stored(app, actioned)
for column in EVENT_COLUMN_RETENTION:
assert getattr(after, column) == getattr(before, column), column
assert after.user_id == submitter.id
assert after.submitted_prompt == "submitted"
assert after.ipaddr is not None
assert not (after.text_redacted or after.ipaddr_redacted or after.anonymized)
assert _stored(app, unactioned).anonymized
@pytest.mark.usefixtures("_no_prior_evidence")
def test_a_trusted_account_at_the_suspicion_threshold_is_not_exempt(app, make_api_user) -> None:
"""``User.is_suspicious`` never reports a trusted account, so its events and worker reports follow every window."""
from horde.classes.base.user import User
from horde.flask import db
trusted = make_api_user(trusted=True)
_add_suspicions(app, trusted.id, User.SUSPICION_THRESHOLD)
with app.app_context():
user = db.session.get(User, trusted.id)
assert len(user.suspicions) >= User.SUSPICION_THRESHOLD
assert not user.is_suspicious()
event = _record_aged(app, user_id=trusted.id, age_days=400)
worker_id = _synthetic_worker(app, make_api_user().id, "synthetic-report-worker")
report = _record_problem_job(app, user_id=trusted.id, worker_id=worker_id, ipaddr=_unique_ip(), age_days=400)
with app.app_context():
assert apply_evidence_retention(_policy(text=10, ipaddr=10, ceiling=30)) == _pass_result(
anonymized=1,
problem_jobs_deleted=1,
)
assert db.session.get(UserProblemJobs, report) is None
assert _stored(app, event).anonymized
@pytest.mark.usefixtures("_no_prior_evidence")
@pytest.mark.parametrize("action", ["flagged", "suspicious"])
def test_worker_reports_of_an_account_under_moderation_action_survive_the_ceiling(app, make_api_user, action: str) -> None:
"""A worker report record of a flagged or suspicious account keeps its address and row past both windows."""
from horde.flask import db
submitter, bystander, worker_owner = make_api_user(), make_api_user(), make_api_user()
worker_id = _synthetic_worker(app, worker_owner.id, "synthetic-report-worker")
ipaddr = _unique_ip()
past_window = _record_problem_job(app, user_id=submitter.id, worker_id=worker_id, ipaddr=ipaddr, age_days=15)
past_ceiling = _record_problem_job(app, user_id=submitter.id, worker_id=worker_id, ipaddr=ipaddr, age_days=400)
unactioned = _record_problem_job(app, user_id=bystander.id, worker_id=worker_id, ipaddr=_unique_ip(), age_days=400)
if action == "flagged":
_flag(app, submitter.id)
else:
_make_suspicious(app, submitter.id)
policy = _policy(text=None, ipaddr=10, ceiling=30)
with app.app_context():
assert apply_evidence_retention(policy) == _pass_result(problem_jobs_deleted=1)
assert apply_evidence_retention(policy) == _pass_result()
for record in (past_window, past_ceiling):
stored = db.session.get(UserProblemJobs, record)
assert stored is not None
assert (stored.user_id, stored.ipaddr) == (submitter.id, ipaddr)
assert db.session.get(UserProblemJobs, unactioned) is None
def _listed_ids(app, **filters: Any) -> set[int]:
"""Return the IDs of the events the filters select; an ``ipaddr`` filter is parsed as the API parses it."""
if "ipaddr" in filters:
filters["ip_subject"] = CounterMeasures.parse_ip_subject(filters.pop("ipaddr"))
with app.app_context():
return {event["id"] for event in get_prompt_events(limit=100, **filters)["events"]}
@pytest.mark.usefixtures("_no_prior_evidence")
def test_address_redaction_keeps_events_in_address_filters(app, make_api_user) -> None:
"""Past the address window the address, and another address in its IPv6 /64, still find the event by its pseudonym."""
submitter = make_api_user()
worker_id = str(uuid4())
proxied_account = f"proxied-{uuid4()}"
common = {
"user_id": submitter.id,
"age_days": 15,
"worker_id": worker_id,
"proxied_account": proxied_account,
}
ipv4 = _unique_ip()
ipv4_event = _record_aged(app, ipaddr=ipv4, **common)
ipv6_event = _record_aged(app, ipaddr="2001:db8:5:6::1", **common)
ipv6_sibling = "2001:db8:5:6::ffff"
now = datetime.utcnow()
retained_filters: list[dict[str, Any]] = [
{"user_id": submitter.id},
{"proxied_account": proxied_account},
{"worker_id": worker_id},
{"since": now - timedelta(days=16), "until": now - timedelta(days=14)},
]
assert _listed_ids(app, ipaddr=ipv4) == {ipv4_event}
assert _listed_ids(app, ipaddr=ipv6_sibling) == {ipv6_event}
with app.app_context():
assert apply_evidence_retention(_policy(text=None, ipaddr=10, ceiling=30)).ipaddr_redacted == 2
assert _stored(app, ipv4_event).ipaddr is None
assert _listed_ids(app, ipaddr=ipv4) == {ipv4_event}
assert _listed_ids(app, ipaddr=f"::ffff:{ipv4}") == {ipv4_event}
assert _listed_ids(app, ipaddr=ipv6_sibling) == {ipv6_event}
assert _listed_ids(app, ipaddr="2001:db8:5:6::/64") == {ipv6_event}
for filters in retained_filters:
assert _listed_ids(app, **filters) == {ipv4_event, ipv6_event}, filters
@pytest.mark.usefixtures("_no_prior_evidence")
def test_events_without_a_pseudonym_match_by_address_until_it_is_removed(app, make_api_user, monkeypatch) -> None:
"""An event captured without a usable secret has no pseudonym, so only its address matches the address filter."""
from horde import countermeasures
submitter = make_api_user()
ipaddr = _unique_ip()
monkeypatch.setattr(countermeasures, "IP_SUBJECT_KEY_SECRET", None)
unkeyed = _record_aged(app, user_id=submitter.id, age_days=15, ipaddr=ipaddr)
monkeypatch.setattr(countermeasures, "IP_SUBJECT_KEY_SECRET", TEST_IP_SUBJECT_SECRET)
keyed = _record_aged(app, user_id=submitter.id, age_days=1, ipaddr=ipaddr)
assert _stored(app, unkeyed).ip_subject_key is None
assert _listed_ids(app, ipaddr=ipaddr) == {unkeyed, keyed}
with app.app_context():
assert apply_evidence_retention(_policy(text=None, ipaddr=10, ceiling=30)).ipaddr_redacted == 1
assert _listed_ids(app, ipaddr=ipaddr) == {keyed}
@pytest.mark.usefixtures("_no_prior_evidence")
def test_anonymization_removes_events_from_subject_filters_only(app, make_api_user) -> None:
"""Past the ceiling the account and proxied-account filters no longer find the event; worker and time filters do."""
submitter = make_api_user()
worker_id = str(uuid4())
proxied_account = f"proxied-{uuid4()}"
expired = _record_aged(app, user_id=submitter.id, age_days=31, worker_id=worker_id, proxied_account=proxied_account)
now = datetime.utcnow()
subject_filters: list[dict[str, Any]] = [
{"user_id": submitter.id},
{"proxied_account": proxied_account},
]
retained_filters: list[dict[str, Any]] = [
{"worker_id": worker_id},
{"since": now - timedelta(days=32)},
{"until": now - timedelta(days=30)},
]
for filters in subject_filters + retained_filters:
assert _listed_ids(app, **filters) == {expired}, filters
with app.app_context():
assert apply_evidence_retention(_policy(text=None, ipaddr=None, ceiling=30)).anonymized == 1
for filters in subject_filters:
assert _listed_ids(app, **filters) == set(), filters
for filters in retained_filters:
assert _listed_ids(app, **filters) == {expired}, filters
@pytest.mark.usefixtures("_no_prior_evidence")
def test_none_windows_keep_text_and_address_until_the_ceiling(app, make_api_user) -> None:
submitter = make_api_user()
@@ -733,6 +1167,32 @@ def test_anonymization_changes_one_batch_oldest_first(app, make_api_user) -> Non
assert apply_evidence_retention(policy, batch_size=2).total == 0
def test_wipe_keeps_the_accounts_evidence_and_notes(client, app, api_key, make_api_user) -> None:
"""Wiping an account leaves its live and anonymized events, their notes and their account in place."""
from horde.classes.base.user import User
from horde.flask import db
wiped, other = make_api_user(), make_api_user()
anonymized_event = _record_aged(app, user_id=wiped.id, age_days=31)
with app.app_context():
apply_evidence_retention(_policy(text=None, ipaddr=None, ceiling=30))
assert _stored(app, anonymized_event).anonymized
wiped_events = _seed(app, user_id=wiped.id, count=2)
(other_event,) = _seed(app, user_id=other.id)
note = client.post(f"{EVENTS_URL}/{wiped_events[0]}/notes", json={"note": "about the account"}, headers={"apikey": api_key})
assert note.status_code == 201, note.get_json()
with app.app_context():
db.session.get(User, wiped.id).wipe()
db.session.expire_all()
assert db.session.get(User, wiped.id).is_wiped
for event_id in wiped_events:
assert db.session.get(PromptModerationEvent, event_id).user_id == wiped.id
assert db.session.get(PromptModerationNote, note.get_json()["id"]) is not None
assert db.session.get(PromptModerationEvent, other_event).user_id == other.id
assert db.session.get(PromptModerationEvent, anonymized_event).anonymized
assert set(wiped_events) <= _listed_ids(app, user_id=wiped.id)
def _record_problem_job(
app,
*,
@@ -906,3 +1366,88 @@ def test_database_failure_does_not_accept_rejected_prompt(client, app, make_api_
)
assert response.status_code == 400, response.get_json()
assert errors == ["Prompt moderation evidence write failed (OperationalError)"]
def test_prompt_event_pagination_filters_and_invalid_notes(client, app, api_key, make_api_user) -> None:
submitter = make_api_user()
ipaddr = _unique_ip()
identifiers = _seed(app, user_id=submitter.id, count=3, proxied_account="proxied-listing", ipaddr=ipaddr)
headers = {"apikey": api_key}
query = {"user_id": submitter.id, "limit": 2}
first = client.get(EVENTS_URL, query_string=query, headers=headers).get_json()
assert [event["id"] for event in first["events"]] == identifiers[:0:-1]
second = client.get(EVENTS_URL, query_string={**query, "before_id": first["next_cursor"]}, headers=headers).get_json()
assert [event["id"] for event in second["events"]] == identifiers[:1]
assert second["next_cursor"] is None
by_address = client.get(EVENTS_URL, query_string={"ipaddr": ipaddr}, headers=headers).get_json()
assert [event["id"] for event in by_address["events"]] == identifiers[::-1]
subject_key = CounterMeasures.ip_subject_key(ipaddr)
assert {event["ip_subject_key"] for event in by_address["events"]} == {subject_key}
by_key = client.get(EVENTS_URL, query_string={"ip_subject_key": subject_key}, headers=headers).get_json()
assert [event["id"] for event in by_key["events"]] == identifiers[::-1]
by_event = client.get(EVENTS_URL, query_string={"event_id": identifiers[1]}, headers=headers).get_json()
assert [event["id"] for event in by_event["events"]] == [identifiers[1]]
blank_address = client.get(EVENTS_URL, query_string={**query, "ipaddr": " "}, headers=headers).get_json()
assert [event["id"] for event in blank_address["events"]] == identifiers[:0:-1]
by_proxied = client.get(
EVENTS_URL,
query_string={"user_id": submitter.id, "proxied_account": "proxied-listing"},
headers=headers,
).get_json()
assert len(by_proxied["events"]) == 3
future = client.get(EVENTS_URL, query_string={**query, "since": "2100-01-01T00:00:00Z"}, headers=headers)
assert future.get_json()["events"] == []
for invalid, rc in (
({"limit": 101}, "InvalidOperationsLimit"),
({"before_id": 0}, "InvalidOperationsCursor"),
({"since": "2026-01-01"}, "InvalidModerationTimeRange"),
({"since": "2026-01-02T00:00:00Z", "until": "2026-01-01T00:00:00Z"}, "InvalidModerationTimeRange"),
({"ipaddr": "not an address"}, "InvalidModerationAddressFilter"),
({"ipaddr": "198.51.100.0/24"}, "InvalidModerationAddressFilter"),
({"ipaddr": "2001:db8::/48"}, "InvalidModerationAddressFilter"),
({"ip_subject_key": "not a pseudonym"}, "InvalidModerationAddressFilter"),
({"ip_subject_key": "A" * 64}, "InvalidModerationAddressFilter"),
({"event_id": 0}, "InvalidModerationEventID"),
):
response = client.get(EVENTS_URL, query_string=invalid, headers=headers)
assert response.status_code == 400, invalid
assert response.get_json()["rc"] == rc, invalid
notes_url = f"{EVENTS_URL}/{identifiers[0]}/notes"
for invalid in ({}, {"note": None}, {"note": ""}, {"note": " "}, {"note": "x" * 2001}, {"note": "nul \x00"}):
response = client.post(notes_url, json=invalid, headers=headers)
assert response.status_code == 400, invalid
assert response.get_json()["rc"] == "InvalidModerationNote", invalid
missing = client.post(f"{EVENTS_URL}/0/notes", json={"note": "gone"}, headers=headers)
assert missing.status_code == 404
assert missing.get_json()["rc"] == "ModerationEventNotFound"
def test_moderator_limits_count_per_key(client, app, api_key, make_api_user) -> None:
from horde.limiter import limiter
other = make_api_user()
limiter.enabled = True
try:
headers = {"apikey": api_key}
for _ in range(120):
assert client.get(EVENTS_URL, query_string={"limit": 1}, headers=headers).status_code == 200
limited = client.get(EVENTS_URL, query_string={"limit": 1}, headers={**headers, "Origin": "http://localhost:4277"})
assert limited.status_code == 429
# A cross-origin client can read how long to wait.
assert limited.headers["Access-Control-Allow-Origin"] == "*"
exposed = {name.strip() for name in limited.headers["Access-Control-Expose-Headers"].split(",")}
assert {"Retry-After", "X-RateLimit-Reset"} <= exposed
assert int(limited.headers["Retry-After"]) > 0
# Another key has its own count.
assert client.get(EVENTS_URL, headers={"apikey": other.api_key}).status_code == 403
# Notes on every event draw on one write budget: the count follows the key and the resource, not the event.
first, second = _seed(app, user_id=other.id, count=2)
for index in range(30):
event_id = first if index % 2 else second
response = client.post(f"{EVENTS_URL}/{event_id}/notes", json={"note": "limit"}, headers=headers)
assert response.status_code == 201, response.get_json()
for event_id in (first, second):
assert client.post(f"{EVENTS_URL}/{event_id}/notes", json={"note": "limit"}, headers=headers).status_code == 429
finally:
limiter.enabled = False
limiter.reset()