feat(assets): manual decommission — the state "Show decommissioned" filtered for
The "Show decommissioned" checkbox sent include_inactive=true and the backend
filter honoured it correctly, but no code path ever wrote status=DECOMMISSIONED:
the syncs only set INACTIVE/ACTIVE, the edit modal had no status field, and
Delete hard-removes the asset with its findings (CASCADE) instead of retiring
it. The filter had no state to filter for, so the checkbox looked dead.
The PUT endpoint did accept `status`, but wrote it with a bare setattr — the
transition reached the audit log as a generic ASSET_UPDATED instead of the
ASSET_DEACTIVATED / ASSET_REACTIVATED entry every sync-driven transition gets.
- Lifecycle Status select in the asset edit modal (edit only, editor+)
- PUT /assets/{id} routes status through apply_status() so the trail names the
transition, with force=True: the operator-final guard is there to stop a sync
from reviving a retired asset, not the operator who retired it
- tests/test_asset_decommission.py pins audit event, default filter and undo
No UI button for the lifecycle reconcile — it stays the nightly 04:15 job plus
POST /assets/reconcile-lifecycle (editor+). The tester text listing a
"Lifecycle abgleichen" button described a control that never existed.
This commit is contained in:
@@ -832,7 +832,15 @@ async def update_asset(
|
||||
|
||||
# Update nur gesetzte Felder
|
||||
import json as _json
|
||||
# Lifecycle transitions do not go through setattr: apply_status() writes the
|
||||
# dedicated ASSET_DEACTIVATED / ASSET_REACTIVATED audit entry, so a manual
|
||||
# decommission reads the same way in the trail as a sync-driven one.
|
||||
status_change: Optional[AssetStatus] = None
|
||||
for field, value in update_data.model_dump(exclude_unset=True).items():
|
||||
if field == "status":
|
||||
if value is not None:
|
||||
status_change = value if isinstance(value, AssetStatus) else AssetStatus(value)
|
||||
continue
|
||||
if field == "compliance_frameworks":
|
||||
# Stored as TEXT JSON list
|
||||
value = _json.dumps(value or []) if value is not None else None
|
||||
@@ -842,6 +850,16 @@ async def update_asset(
|
||||
raise HTTPException(400, "criticality must be one of: low, normal, high, critical")
|
||||
setattr(asset, field, value)
|
||||
|
||||
if status_change is not None:
|
||||
from app.services.asset_lifecycle import apply_status
|
||||
# force=True: the operator-final guard exists to stop a sync from
|
||||
# reviving a retired asset, not the operator who retired it.
|
||||
apply_status(
|
||||
db, asset, status_change,
|
||||
f"manual status change by {current_user.username}",
|
||||
force=True,
|
||||
)
|
||||
|
||||
db.commit()
|
||||
db.refresh(asset)
|
||||
|
||||
|
||||
@@ -164,7 +164,13 @@ def _liveness_stamp(asset: Asset) -> Optional[datetime]:
|
||||
return max(stamps) if stamps else None
|
||||
|
||||
|
||||
def apply_status(db: Session, asset: Asset, new: AssetStatus, reason: str) -> bool:
|
||||
def apply_status(
|
||||
db: Session,
|
||||
asset: Asset,
|
||||
new: AssetStatus,
|
||||
reason: str,
|
||||
force: bool = False,
|
||||
) -> bool:
|
||||
"""Set `asset.status` and audit-log the transition. No-op if unchanged.
|
||||
|
||||
Every sync that writes a status must go through here. Writing
|
||||
@@ -174,10 +180,12 @@ def apply_status(db: Session, asset: Asset, new: AssetStatus, reason: str) -> bo
|
||||
trail must not do (field report 2026-08-18).
|
||||
|
||||
DECOMMISSIONED is operator-final: a sync never revives it, mirroring both
|
||||
reconcile functions in this module.
|
||||
reconcile functions in this module. `force=True` lifts that guard for the
|
||||
one caller it was never meant to stop — the human who retired the asset
|
||||
and now wants it back (PUT /assets/{id}). Syncs must never pass it.
|
||||
"""
|
||||
old = asset.status.value if hasattr(asset.status, "value") else str(asset.status)
|
||||
if old == AssetStatus.DECOMMISSIONED.value:
|
||||
if old == AssetStatus.DECOMMISSIONED.value and not force:
|
||||
return False
|
||||
if old == new.value:
|
||||
return False
|
||||
|
||||
@@ -50,6 +50,7 @@ export default function AssetsPage() {
|
||||
description: string;
|
||||
policy_id?: number | null;
|
||||
criticality: 'low' | 'normal' | 'high' | 'critical';
|
||||
status?: 'active' | 'inactive' | 'decommissioned';
|
||||
}>({
|
||||
hostname: '',
|
||||
ip_address: '',
|
||||
@@ -148,6 +149,7 @@ export default function AssetsPage() {
|
||||
description: '',
|
||||
policy_id: undefined,
|
||||
criticality: 'normal',
|
||||
status: undefined,
|
||||
});
|
||||
setEditingId(null);
|
||||
setFormVmwareBuild(null);
|
||||
@@ -206,6 +208,7 @@ export default function AssetsPage() {
|
||||
description: asset.description || '',
|
||||
policy_id: asset.policy_id,
|
||||
criticality: (asset as any).criticality || 'normal',
|
||||
status: asset.status,
|
||||
});
|
||||
setFormVmwareBuild(asset.vmware_build || null);
|
||||
setIsModalOpen(true);
|
||||
@@ -607,6 +610,27 @@ export default function AssetsPage() {
|
||||
<option value="critical">Critical (×1.5)</option>
|
||||
</select>
|
||||
</div>
|
||||
{editingId && (
|
||||
<div>
|
||||
<label className="block text-sm font-medium text-gray-700 font-mono">
|
||||
Lifecycle Status
|
||||
</label>
|
||||
<select
|
||||
value={formData.status || 'active'}
|
||||
onChange={(e) => setFormData({ ...formData, status: e.target.value as any })}
|
||||
className="mt-1 block w-full rounded-md border-gray-300 shadow-sm focus:border-truevuln-blue focus:ring-truevuln-blue sm:text-sm font-mono"
|
||||
>
|
||||
<option value="active">Active — reported by a source</option>
|
||||
<option value="inactive">Inactive — not seen inside the grace window</option>
|
||||
<option value="decommissioned">Decommissioned — operator-retired</option>
|
||||
</select>
|
||||
<p className="mt-1 text-xs text-gray-500 font-mono">
|
||||
DECOMMISSIONED hides the asset and its findings from every default view
|
||||
(tick "Show decommissioned" to see them) and no sync revives it.
|
||||
History and audit trail are kept — unlike Delete, which cascades them away.
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
<div>
|
||||
<label className="block text-sm font-medium text-gray-700 font-mono">Assign Policy</label>
|
||||
<select
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
"""Manual decommission: audited, filtered, reversible — run: python tests/test_asset_decommission.py
|
||||
|
||||
The "Show decommissioned" checkbox filtered on a status no code path ever
|
||||
wrote: syncs only ever set INACTIVE/ACTIVE, the edit modal had no status
|
||||
field, and Delete hard-removes the asset (CASCADE) instead of retiring it.
|
||||
The PUT endpoint accepted `status` but wrote it with a bare setattr, so the
|
||||
transition reached the audit log as a generic ASSET_UPDATED.
|
||||
|
||||
Pinned here: the manual transition is audited as a lifecycle event, the
|
||||
default asset filter hides it, and the operator (not a sync) can undo it.
|
||||
"""
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "test-key-test-key-test-key-test-key")
|
||||
|
||||
from sqlalchemy import create_engine # noqa: E402
|
||||
from sqlalchemy.orm import sessionmaker # noqa: E402
|
||||
|
||||
import app.models # noqa: F401,E402 (registers every mapper)
|
||||
from app.models.base import Base # noqa: E402
|
||||
from app.models.asset import Asset, AssetSource, AssetStatus # noqa: E402
|
||||
from app.models.audit_log import AuditLog # noqa: E402
|
||||
from app.models.user import User # noqa: E402
|
||||
from app.routers.assets import ( # noqa: E402
|
||||
AssetUpdateRequest,
|
||||
_apply_asset_filters,
|
||||
update_asset,
|
||||
)
|
||||
from app.services.asset_lifecycle import apply_status # noqa: E402
|
||||
|
||||
|
||||
def _db():
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
db = sessionmaker(bind=engine)()
|
||||
for name in ("a-active", "b-inactive", "c-target"):
|
||||
db.add(Asset(hostname=name, ip_address="10.0.0.1", source=AssetSource.MANUAL,
|
||||
status=AssetStatus.INACTIVE if name == "b-inactive" else AssetStatus.ACTIVE))
|
||||
db.add(User(id=1, username="op", email="op@example.com", password_hash="x"))
|
||||
db.commit()
|
||||
return db
|
||||
|
||||
|
||||
def _put(db, asset, **fields):
|
||||
user = db.query(User).first()
|
||||
return asyncio.run(update_asset(asset.id, AssetUpdateRequest(**fields), db, user))
|
||||
|
||||
|
||||
def _names(db, **kw):
|
||||
return sorted(a.hostname for a in _apply_asset_filters(db.query(Asset), db, **kw).all())
|
||||
|
||||
|
||||
def test_manual_decommission_is_audited_and_filtered():
|
||||
db = _db()
|
||||
target = db.query(Asset).filter(Asset.hostname == "c-target").first()
|
||||
|
||||
_put(db, target, status=AssetStatus.DECOMMISSIONED, description="retired")
|
||||
db.refresh(target)
|
||||
assert target.status == AssetStatus.DECOMMISSIONED, target.status
|
||||
assert target.description == "retired", "other fields must still be written"
|
||||
|
||||
events = [a.event_type.value if hasattr(a.event_type, "value") else str(a.event_type)
|
||||
for a in db.query(AuditLog).all()]
|
||||
assert "ASSET_DEACTIVATED" in events, f"lifecycle transition not audited: {events}"
|
||||
|
||||
# INACTIVE stays visible; only the operator-retired asset drops out.
|
||||
assert _names(db) == ["a-active", "b-inactive"], _names(db)
|
||||
assert _names(db, include_inactive=True) == ["a-active", "b-inactive", "c-target"]
|
||||
|
||||
|
||||
def test_operator_can_undo_but_a_sync_cannot():
|
||||
db = _db()
|
||||
target = db.query(Asset).filter(Asset.hostname == "c-target").first()
|
||||
_put(db, target, status=AssetStatus.DECOMMISSIONED)
|
||||
|
||||
# A sync (no force) must never revive an operator-retired asset.
|
||||
assert apply_status(db, target, AssetStatus.ACTIVE, "seen again by a sync") is False
|
||||
assert target.status == AssetStatus.DECOMMISSIONED
|
||||
|
||||
# The operator who retired it can put it back.
|
||||
_put(db, target, status=AssetStatus.ACTIVE)
|
||||
db.refresh(target)
|
||||
assert target.status == AssetStatus.ACTIVE, target.status
|
||||
assert _names(db) == ["a-active", "b-inactive", "c-target"]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_manual_decommission_is_audited_and_filtered()
|
||||
test_operator_can_undo_but_a_sync_cannot()
|
||||
print("OK — manual decommission audited, filtered, operator-reversible")
|
||||
Reference in New Issue
Block a user