"""Two-step delete guard: preview -> confirm -> execute.
Every delete operation requires:
1. guard_delete_task / guard_delete_project -> returns preview + confirm_token
2. guard_confirm_delete -> validates token, executes, logs
"""
from __future__ import annotations
import hashlib
import secrets
import time
from typing import Any
import config
import asana_client
import audit_logger
from audit_logger import AuditTimer
_pending_deletes: dict[str, dict[str, Any]] = {}
def _generate_token(entity_type: str, gid: str) -> str:
raw = secrets.token_urlsafe(32)
token_hash = hashlib.sha256(raw.encode()).hexdigest()
_pending_deletes[token_hash] = {
"entity_type": entity_type,
"gid": gid,
"created_at": time.monotonic(),
}
return raw
def _validate_token(raw_token: str, entity_type: str, gid: str) -> bool:
token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
entry = _pending_deletes.pop(token_hash, None)
if entry is None:
return False
elapsed = time.monotonic() - entry["created_at"]
if elapsed > config.CONFIRM_TOKEN_TTL_SECONDS:
return False
return entry["entity_type"] == entity_type and entry["gid"] == gid
def _cleanup_expired() -> None:
now = time.monotonic()
expired = [
k for k, v in _pending_deletes.items()
if now - v["created_at"] > config.CONFIRM_TOKEN_TTL_SECONDS * 2
]
for k in expired:
_pending_deletes.pop(k, None)
def preview_delete_task(task_gid: str) -> dict[str, Any]:
"""Fetch task details and issue a confirm token."""
_cleanup_expired()
task = asana_client.get_task(task_gid)
token = _generate_token("task", task_gid)
preview = {
"action": "DELETE_TASK",
"task_gid": task_gid,
"name": task.get("name", ""),
"projects": [p.get("name", "") for p in task.get("projects", [])],
"assignee": (task.get("assignee") or {}).get("name", "unassigned"),
"completed": task.get("completed", False),
"confirm_token": token,
"token_expires_in_seconds": config.CONFIRM_TOKEN_TTL_SECONDS,
"warning": "This action is IRREVERSIBLE. Provide the confirm_token to guard_confirm_delete to proceed.",
}
audit_logger.log_operation(
tool="guard_delete_task",
params={"task_gid": task_gid},
asana_gid=task_gid,
result="preview_issued",
)
return preview
def preview_delete_project(project_gid: str) -> dict[str, Any]:
"""Fetch project details and issue a confirm token."""
_cleanup_expired()
project = asana_client.get_project(project_gid)
token = _generate_token("project", project_gid)
preview = {
"action": "DELETE_PROJECT",
"project_gid": project_gid,
"name": project.get("name", ""),
"owner": (project.get("owner") or {}).get("name", "unknown"),
"team": (project.get("team") or {}).get("name", ""),
"confirm_token": token,
"token_expires_in_seconds": config.CONFIRM_TOKEN_TTL_SECONDS,
"warning": "This will DELETE the entire project and all its tasks. Provide the confirm_token to guard_confirm_delete to proceed.",
}
audit_logger.log_operation(
tool="guard_delete_project",
params={"project_gid": project_gid},
asana_gid=project_gid,
result="preview_issued",
)
return preview
def confirm_delete(
entity_type: str,
gid: str,
confirm_token: str,
) -> dict[str, Any]:
"""Validate the token and execute the deletion."""
if entity_type not in ("task", "project"):
return {"error": f"Invalid entity_type: {entity_type}. Must be 'task' or 'project'."}
if not _validate_token(confirm_token, entity_type, gid):
audit_logger.log_operation(
tool="guard_confirm_delete",
params={"entity_type": entity_type, "gid": gid},
asana_gid=gid,
error="invalid_or_expired_token",
)
return {
"error": "Invalid or expired confirm_token. Request a new preview first.",
}
with AuditTimer() as timer:
try:
if entity_type == "task":
asana_client.delete_task(gid)
else:
asana_client.delete_project(gid)
except Exception as exc:
audit_logger.log_operation(
tool="guard_confirm_delete",
params={"entity_type": entity_type, "gid": gid},
asana_gid=gid,
error=str(exc),
duration_ms=timer.elapsed_ms,
)
return {"error": f"Deletion failed: {exc}"}
audit_logger.log_operation(
tool="guard_confirm_delete",
params={"entity_type": entity_type, "gid": gid},
asana_gid=gid,
result="deleted",
duration_ms=timer.elapsed_ms,
)
return {
"status": "deleted",
"entity_type": entity_type,
"gid": gid,
"note": "Deletion is permanent. Asana does not support undo via API.",
}
