69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
"""
|
|
The two periodic jobs the access timeline needs, wired to a schedule.
|
|
|
|
`app/modules/org/services/access_maintenance.py` has had both of these written,
|
|
tested and callable since the C4 work, with a deliberate note that neither was
|
|
scheduled: "choosing when they run is an operational decision, and a task that
|
|
fires on an interval nobody chose is worse than one that has not been scheduled
|
|
yet."
|
|
|
|
That was the right call then and it is the wrong state to leave now. Roles can
|
|
now be granted with an expiry from the admin screens, so a temporary grant is
|
|
something an ordinary administrator will create — and without the sweep the
|
|
timeline shows such a grant appearing and then simply ceasing to apply, with
|
|
nothing marking when. "When did their access end" is the question the log exists
|
|
to answer.
|
|
|
|
**Expiry is enforced regardless of this task.** `PermissionService` and
|
|
`ScopeService` both filter on `expires_at` in SQL, so a lapsed grant confers
|
|
nothing whether or not the sweeper has run. This writes *history*, not
|
|
enforcement; if it never runs, access is still correct and the log is merely
|
|
incomplete.
|
|
|
|
**Only the expiry recorder is scheduled.** `prune_access_log_task` is defined and
|
|
callable and is deliberately left off the beat schedule:
|
|
`ACCESS_LOG_RETENTION_DAYS` defaults to 365, so scheduling it would begin
|
|
deleting year-old access history because of a default nobody chose. Retention is
|
|
usually a contractual number. Set it deliberately, then add the entry in
|
|
`app/tasks/celery_app.py`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from celery import shared_task
|
|
|
|
from app.db.database import get_db_connection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@shared_task
|
|
def record_grant_expiries_task() -> int:
|
|
"""Write one `role_expired` event per grant that has lapsed since the last run."""
|
|
from app.modules.org.services.access_maintenance import record_expiries
|
|
|
|
for db in get_db_connection():
|
|
written = record_expiries(db)
|
|
if written:
|
|
logger.info("Recorded %d grant expiry event(s)", written)
|
|
return written
|
|
return 0
|
|
|
|
|
|
@shared_task
|
|
def prune_access_log_task() -> int:
|
|
"""Delete access events older than `ACCESS_LOG_RETENTION_DAYS`, if set."""
|
|
from app.modules.org.services.access_maintenance import prune_access_log
|
|
|
|
for db in get_db_connection():
|
|
deleted = prune_access_log(db)
|
|
if deleted:
|
|
logger.info("Pruned %d access log row(s)", deleted)
|
|
return deleted
|
|
return 0
|
|
|
|
|
|
__all__ = ["record_grant_expiries_task", "prune_access_log_task"]
|