mirror of
https://github.com/yogeshojha/rengine.git
synced 2026-10-20 07:11:28 +02:00
Yogesh OjhaandClaude Opus 5.5
30f2c47ed9
Only a capability stage's success covers a dimension; a support stage covers through the rows it wrote. A cancelled run that finished takeover and zone_transfer hid the previous run's findings on every page. The targets list reads findings from each target's vulnerability covering scan, and both strips gain a medium chip. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
"""A focused rescan is evidence, not a census — every surface rollup filters on this."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from functools import lru_cache
|
|
|
|
from sqlalchemy import exists, select
|
|
|
|
from shared.definitions.surface import SURFACE_KINDS, SURFACE_ORDER
|
|
from shared.enums.scan import ScanActivityStatus, ScanScope, StageRole
|
|
from shared.models.scan import Scan
|
|
from shared.models.scan_activity import ScanActivity
|
|
|
|
|
|
def census_only(model=Scan):
|
|
"""Runs that describe a target's whole surface."""
|
|
return model.scope == ScanScope.FULL.value
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def producing_stages() -> dict[str, frozenset[str]]:
|
|
"""Dimension -> every stage that can write its rows."""
|
|
from stages.registry import stages # noqa: PLC0415
|
|
|
|
out: dict[str, set[str]] = {key: set() for key in SURFACE_ORDER}
|
|
for spec in stages():
|
|
for key, kinds in SURFACE_KINDS.items():
|
|
if spec.produces & kinds:
|
|
out[key].add(spec.name)
|
|
return {key: frozenset(names) for key, names in out.items()}
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def covering_stages() -> dict[str, frozenset[str]]:
|
|
"""Dimension -> the capability stages whose success means the dimension was scanned."""
|
|
from stages.registry import stages # noqa: PLC0415
|
|
|
|
capability = {s.name for s in stages() if s.role == StageRole.CAPABILITY.value}
|
|
return {
|
|
key: (names & capability) or names for key, names in producing_stages().items()
|
|
}
|
|
|
|
|
|
def covers(model, dimension: str, scan=Scan):
|
|
"""The scan wrote rows for the dimension, or a producing stage succeeded."""
|
|
return exists(select(1).where(model.scan_id == scan.id)) | exists(
|
|
select(1).where(
|
|
ScanActivity.scan_id == scan.id,
|
|
ScanActivity.status == ScanActivityStatus.SUCCESS.value,
|
|
ScanActivity.name.in_(covering_stages()[dimension]),
|
|
)
|
|
)
|