Audit

dagster_authkit.utils.audit

Structured JSON audit logging to stdout.

All security-relevant events (logins, access control, rate limit violations, password changes) are emitted as single-line JSON objects suitable for ingestion by log aggregators (Splunk, Datadog, ELK).

Public API: - log_audit_event — generic audit event - log_login_attempt — login success/failure - log_logout — session termination - log_access_control — RBAC allow/deny - log_rate_limit_violation — brute-force protection trigger - log_password_changed — credential rotation

AuditLogger

Thread-safe audit event emitter.

All events are serialised as single-line JSON to stdout via a dedicated dagster_authkit.audit logger. The service and env fields are automatically stamped on every event.

Usage::

audit = AuditLogger()
audit.login_attempt("admin", True, ip="10.0.0.1")

__init__

__init__()

Initialise with service name and environment tag.

log_event

log_event(event_type, performed_by, **kwargs)

Emit a generic audit event.

Parameters:

Name	Type	Description	Default
event_type	str	Category label (e.g. USER_CREATED, LOGOUT).	required
performed_by	str	Actor that triggered the event (username or system).	required
**kwargs		Arbitrary context fields serialised into the JSON payload.	{}

login_attempt

login_attempt(username, success, ip=None, reason=None)

Log a login attempt.

Parameters:

Name	Type	Description	Default
username		Authenticated username (or attempted username on failure).	required
success		True if authentication succeeded.	required
ip		Client IP address.	None
reason		Failure reason (e.g. INVALID_CREDENTIALS, RATE_LIMIT).	None

access_control

access_control(username, action, resource, allowed, roles=None, reason=None)

Log an RBAC access control decision.

Parameters:

Name	Type	Description	Default
username		User performing the action.	required
action		Operation attempted (mutation name or REST method).	required
resource		Target resource (URL path or GraphQL field).	required
allowed		True if access was granted.	required
roles		List of roles held by the user.	None
reason		Explanation if denied (e.g. REQUIRES_ADMIN).	None

rate_limit_violation

rate_limit_violation(username, ip=None, attempts=0)

Log a rate-limit violation.

Parameters:

Name	Type	Description	Default
username		Identifier that triggered the limit.	required
ip		Client IP address.	None
attempts		Number of attempts within the window.	0

log_audit_event

log_audit_event(event_type, performed_by, **kwargs)

Emit a generic audit event (convenience wrapper).

Parameters:

Name	Type	Description	Default
event_type		Category label (e.g. USER_CREATED, SESSION_CREATED).	required
performed_by		Actor identifier (username or system).	required
**kwargs		Additional context fields (target, role, ip, etc.).	{}

log_login_attempt

log_login_attempt(u, s, ip=None, r=None)

Log a login attempt. See AuditLogger.login_attempt.

log_logout

log_logout(u, ip=None)

Log a user logout event.

Parameters:

Name	Type	Description	Default
u		Username that logged out.	required
ip		Client IP address.	None

log_access_control

log_access_control(u, a, res, al, roles=None, r=None)

Log an RBAC access control decision. See AuditLogger.access_control.

log_rate_limit_violation

log_rate_limit_violation(u, ip=None, att=0)

Log a rate-limit violation. See AuditLogger.rate_limit_violation.

log_password_changed

log_password_changed(u, pb, ss=False)

Log a password change event.

Parameters:

Name	Type	Description	Default
u		Username of the target user.	required
pb		Actor who performed the change.	required
ss		True if the user changed their own password (self-service).	False