security(07.2-03): add import time + user_nbf write to admin deactivation handler; promote stubs to PASSED
This commit is contained in:
@@ -23,10 +23,12 @@ Security invariants:
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import time
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
|
from config import settings
|
||||||
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
||||||
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
|
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
|
||||||
from sqlalchemy import func, select, update
|
from sqlalchemy import func, select, update
|
||||||
@@ -351,6 +353,12 @@ async def update_user_status(
|
|||||||
if not body.is_active:
|
if not body.is_active:
|
||||||
# Revoke all refresh tokens on deactivation
|
# Revoke all refresh tokens on deactivation
|
||||||
await revoke_all_refresh_tokens(session, user.id)
|
await revoke_all_refresh_tokens(session, user.id)
|
||||||
|
# Revoke any pre-deactivation access tokens still within their TTL (T-7.2-01)
|
||||||
|
await request.app.state.redis.set(
|
||||||
|
f"user_nbf:{user.id}",
|
||||||
|
int(time.time()),
|
||||||
|
ex=settings.access_token_expire_minutes * 60,
|
||||||
|
)
|
||||||
|
|
||||||
session.add(user)
|
session.add(user)
|
||||||
|
|
||||||
|
|||||||
@@ -450,10 +450,11 @@ async def test_delete_user_no_body(admin_client):
|
|||||||
# user_nbf — only deactivation writes it (RESEARCH.md anti-pattern).
|
# user_nbf — only deactivation writes it (RESEARCH.md anti-pattern).
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 2 — user_nbf write not yet added to admin deactivation handler")
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_deactivate_user_writes_user_nbf_to_redis(admin_client):
|
async def test_deactivate_user_writes_user_nbf_to_redis(admin_client):
|
||||||
"""PATCH /api/admin/users/{id}/status {is_active: false} must write user_nbf:{id} to Redis."""
|
"""PATCH /api/admin/users/{id}/status {is_active: false} must write user_nbf:{id} to Redis."""
|
||||||
|
from main import app
|
||||||
|
|
||||||
client, _admin, session = admin_client
|
client, _admin, session = admin_client
|
||||||
target = await make_regular_user(session)
|
target = await make_regular_user(session)
|
||||||
|
|
||||||
@@ -463,11 +464,11 @@ async def test_deactivate_user_writes_user_nbf_to_redis(admin_client):
|
|||||||
)
|
)
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
# Wave 2 will add the user_nbf write; this is the target assertion:
|
nbf_bytes = await app.state.redis.get(f"user_nbf:{target.id}")
|
||||||
assert False, "stub — Wave 2 will fill in: nbf_bytes = await app.state.redis.get(f'user_nbf:{target.id}'); assert nbf_bytes is not None"
|
assert nbf_bytes is not None
|
||||||
|
assert int(nbf_bytes.decode() if isinstance(nbf_bytes, (bytes, bytearray)) else nbf_bytes) > 0
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 2 — activation must NOT write user_nbf (invariant guard)")
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_activate_user_does_not_write_user_nbf(admin_client):
|
async def test_activate_user_does_not_write_user_nbf(admin_client):
|
||||||
"""PATCH /api/admin/users/{id}/status {is_active: true} must NOT write user_nbf:{id} to Redis.
|
"""PATCH /api/admin/users/{id}/status {is_active: true} must NOT write user_nbf:{id} to Redis.
|
||||||
@@ -498,5 +499,5 @@ async def test_activate_user_does_not_write_user_nbf(admin_client):
|
|||||||
)
|
)
|
||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
|
|
||||||
# Wave 2 must preserve this invariant; this is the target assertion:
|
nbf_bytes = await app.state.redis.get(f"user_nbf:{target.id}")
|
||||||
assert False, "stub — Wave 2 will fill in: nbf_bytes = await app.state.redis.get(f'user_nbf:{target.id}'); assert nbf_bytes is None"
|
assert nbf_bytes is None, "activation must NOT write user_nbf (anti-pattern guard)"
|
||||||
|
|||||||
Reference in New Issue
Block a user