diff --git a/docs/dr_egeria_manual.md b/docs/dr_egeria_manual.md index 5254452d..d39fb790 100644 --- a/docs/dr_egeria_manual.md +++ b/docs/dr_egeria_manual.md @@ -271,6 +271,7 @@ Dr.Egeria organizes its commands into "families," each corresponding to a specif - `Create`/`Update Statement of Applicability`: an AuthoredReferenceable with `Scope`, `Review Date`, `Authors` and `Content Status` (set APPROVED before an audit). `Certification Standard` links the certification type it is assessed against (GovernedBy); `Scoped By` links the organization, site or system it covers (ScopedBy). Leaving either out leaves its links alone; `Merge Update: false` also removes links the command no longer lists. - `Link`/`Unlink Control Applicability`: assess a control (any governance definition, typically a GovernanceControl) in a statement — `Is Applicable`, `Justification`, `Inclusion Reasons`, `Implementation Status` (NOT_STARTED, PLANNED, PARTIALLY_IMPLEMENTED, IMPLEMENTED, OTHER), `Implementation Description`, `Review Date`, and `Label` for the control's number in the standard. There is one relationship per control, and re-running Link updates it. Egeria's attach replaces the relationship's properties wholesale, so the command starts from the existing ones: re-linking with just a new status keeps the justification and the rest (`Merge Update: false` replaces them). - Reports: `Statement-of-Applicability-Controls` (one row per control — use `LIST` for a markdown table, `DICT` for a spreadsheet) and `Statement-of-Applicability-Implementation-Summary` (applicable controls counted by implementation status; `BAR`/`PIE` for a chart). Both take `soa_guid`, which accepts the statement's GUID or qualified name, e.g. `View Report` with `Report Parameters` `soa_guid: StatementOfApplicability::Coco Pharmaceuticals ISMS`. The summary is analytic-function-backed, which `View Report` cannot run yet; use it through `exec_report_spec` or `SecurityOfficer.get_control_implementation_summary`. + - Security testing results (Egeria 6.3, model 0665): each test run is a `SurveyReport`, with one `SecurityFindingAnnotation` (or `VulnerabilityAnnotation` / `PenetrationTestFindingAnnotation`) per finding and a `SecurityPostureAnnotation` summarizing the run. The report specs `Security-Findings` (most severe first; params `report_guid` or `element_guid`, each a GUID or qualified name, and optional `min_severity`), `Penetration-Test-Findings` and `Security-Posture` (one row per run, oldest first; optional `report_qualified_name_prefix` to follow one system's series) run through `View Report`. Egeria returns `cvssScore`, `exploitAvailable`, `patchAvailable` and the posture counts as 0 or false when they were never set, so these reports show such values blank rather than as an answer; new and resolved finding counts of 0 cannot be told apart from unrecorded. Dr.Egeria commands for creating survey reports and findings are not available yet. - **Action Author**: Define governance action process flows — reusable single-step action types and multi-step processes — and wire them to the engines and elements that execute them, without writing code (e.g., `Create Governance Action Process`, `Create Governance Action Process Step`, `Link First Process Step`, `Link Next Process Step`, `Link Action to Action Executor`, `Link Action to Target`). - **Curation**: Apply classifications and relationships to *existing* Referenceable elements after the fact — this family creates no elements of its own, it curates ones created elsewhere. Every command names a `Target Element` (Reference Name) and either a classification level/status or a second element to relate to. - **Classifications** (`Classify`/`Reclassify`/`Declassify`, or `Classify`/`Update`/`Declassify` for the three multi-value ones): Impact, Confidence, Confidentiality, Criticality, Retention, Ownership, Digital Resource Origin, Zone Membership, Security Tags, Data Scope, Governance Expectations, Governance Measurements, Known Duplicate, Consolidated Duplicate, and the 0438 naming-standards markers Class Word, Modifier, Prime Word (added 2026-08-09 once Egeria PR #9166 shipped the backing `glossary-manager` endpoints — routed through `GlossaryManager`, not `ClassificationExplorer`, unlike every other classification here). Also, added 2026-08-21 once an `omvs_audit.py` gap-closure pass shipped their backing REST endpoints (all confirmed against a live 6.2-SNAPSHOT server): the 10 governance-point classifications (0435: Control/Verification/Enforcement/Execution Point, Policy Administration/Decision/Enforcement/Information/Management/Retrieval Point — routed through `GovernanceOfficer`), 6 metamodel markers (0463: Incomplete, ObjectIdentifier, ReferenceData, MobileResource, InstanceMetadata, MetamodelInstance — routed through `ClassificationExplorer`), ProjectKind (routed through `ProjectManager`), CollectionKind (routed through `CollectionManager`), and a Data Sharing Agreement retrofit pair for classifying/declassifying an already-existing Agreement (routed through `DigitalBusiness` — `Create Data Sharing Agreement`, Digital Products family, already sets this at creation time via `initialClassifications`). Also Promise (0010, added 2026-09-30 — `Classify`/`Declassify Promise`, routed through `ClassificationExplorer`): marks an element as a placeholder for a real-world resource that has not yet been delivered, with Deployment Status, Start/Due/Last Review/Completion Time and Additional Properties. A promised element is only visible to lineage requests, so `Declassify Promise` looks up its Target Element and sends the clear request with `forLineage=true` automatically. diff --git a/pyegeria/models/__init__.py b/pyegeria/models/__init__.py index 3c685623..a8f02477 100644 --- a/pyegeria/models/__init__.py +++ b/pyegeria/models/__init__.py @@ -77,6 +77,12 @@ AuthoredReferenceableProperties, SolutionBlueprintProperties, StatementOfApplicabilityProperties, + AnnotationProperties, + ResourceMeasureAnnotationProperties, + SecurityFindingAnnotationProperties, + VulnerabilityAnnotationProperties, + PenetrationTestFindingAnnotationProperties, + SecurityPostureAnnotationProperties, ControlApplicabilityProperties, ControlImplementationStatus, NestedDesignPatternProperties, @@ -174,6 +180,12 @@ "AuthoredReferenceableProperties", "SolutionBlueprintProperties", "StatementOfApplicabilityProperties", + "AnnotationProperties", + "ResourceMeasureAnnotationProperties", + "SecurityFindingAnnotationProperties", + "VulnerabilityAnnotationProperties", + "PenetrationTestFindingAnnotationProperties", + "SecurityPostureAnnotationProperties", "ControlApplicabilityProperties", "ControlImplementationStatus", "NestedDesignPatternProperties", diff --git a/pyegeria/models/models.py b/pyegeria/models/models.py index 38cc8678..4ba50a26 100644 --- a/pyegeria/models/models.py +++ b/pyegeria/models/models.py @@ -223,6 +223,72 @@ class StatementOfApplicabilityProperties(AuthoredReferenceableProperties): review_date: datetime | None = None +class AnnotationProperties(AuthoredReferenceableProperties): + class_: Annotated[Literal["AnnotationProperties"], Field(alias="class")] + annotation_type: str | None = None + summary: str | None = None + sample_size: int | None = None + sample_percent: int | None = None + sampling_method: str | None = None + confidence: int | None = None + units: str | None = None + absolute_uncertainty: int | None = None + relative_uncertainty: int | None = None + expression: str | None = None + explanation: str | None = None + analysis_step: str | None = None + json_properties: str | None = None + user_defined_content_status: str | None = None + + +class ResourceMeasureAnnotationProperties(AnnotationProperties): + class_: Annotated[Literal["ResourceMeasureAnnotationProperties"], Field(alias="class")] + resource_properties: dict[str, str] | None = None + + +class SecurityFindingAnnotationProperties(AnnotationProperties): + """A security finding (Egeria 6.3, model 0665). Severity uses the CVSS ratings.""" + class_: Annotated[Literal["SecurityFindingAnnotationProperties"], Field(alias="class")] + severity: Literal["None", "Low", "Medium", "High", "Critical"] | None = None + cvss_score: float | None = Field(default=None, ge=0.0, le=10.0) + cvss_vector: str | None = None + vulnerability_identifiers: list[str] | None = None + weakness_identifiers: list[str] | None = None + affected_location: str | None = None + evidence: str | None = None + remediation: str | None = None + exploit_available: bool | None = None + + +class VulnerabilityAnnotationProperties(SecurityFindingAnnotationProperties): + """A known vulnerability in the software a resource uses, typically from a scan.""" + class_: Annotated[Literal["VulnerabilityAnnotationProperties"], Field(alias="class")] + package_url: str | None = Field(default=None, alias="packageURL") + installed_version: str | None = None + fixed_version: str | None = None + patch_available: bool | None = None + + +class PenetrationTestFindingAnnotationProperties(SecurityFindingAnnotationProperties): + """A weakness a penetration test was able to exploit.""" + class_: Annotated[Literal["PenetrationTestFindingAnnotationProperties"], Field(alias="class")] + attack_vector: str | None = None + attack_techniques: list[str] | None = None + privileges_gained: str | None = None + compromised_resources: list[str] | None = None + reproduction_steps: str | None = None + + +class SecurityPostureAnnotationProperties(ResourceMeasureAnnotationProperties): + """The summary of one security test run.""" + class_: Annotated[Literal["SecurityPostureAnnotationProperties"], Field(alias="class")] + finding_counts_by_severity: dict[str, int] | None = None + resources_in_scope_count: int | None = None + resources_tested_count: int | None = None + new_finding_count: int | None = None + resolved_finding_count: int | None = None + + class ControlApplicabilityProperties(PyegeriaModel): class_: Annotated[Literal["ControlApplicabilityProperties"], Field(alias="class")] # Egeria treats a control as applicable unless this is false. diff --git a/pyegeria/omvs/data_discovery.py b/pyegeria/omvs/data_discovery.py index ceda2e54..cde1bab5 100644 --- a/pyegeria/omvs/data_discovery.py +++ b/pyegeria/omvs/data_discovery.py @@ -6,11 +6,11 @@ """ import asyncio -from typing import Annotated, Literal, Optional - -from pydantic import Field +import uuid +from typing import Optional from pyegeria.core._server_client import ServerClient +from pyegeria.core._globals import NO_ELEMENTS_FOUND from pyegeria.models import ( NewElementRequestBody, DeleteElementRequestBody, @@ -22,7 +22,7 @@ GetRequestBody, NewRelationshipRequestBody, DeleteRelationshipRequestBody, - ReferenceableProperties, + AnnotationProperties, ) from pyegeria.view.output_formatter import ( generate_output, @@ -32,15 +32,6 @@ from pyegeria.core.utils import dynamic_catch -class AnnotationProperties(ReferenceableProperties): - class_: Annotated[Literal["AnnotationProperties"], Field(alias="class")] - annotation_type: Optional[str] = None - summary: Optional[str] = None - confidence_level: Optional[int] = None - expression: Optional[str] = None - explanation: Optional[str] = None - analysis_step: Optional[str] = None - json_properties: Optional[str] = None class DataDiscovery(ServerClient): @@ -1684,6 +1675,248 @@ def get_annotation_by_guid( ) ) + # + # Security findings and posture (Egeria 6.3, model 0665). A security test run is a SurveyReport; + # each finding is a SecurityFindingAnnotation (or its Vulnerability / PenetrationTestFinding subtypes) + # and the run's summary is a SecurityPostureAnnotation, all linked to the report with ReportedAnnotation. + # + + SECURITY_FINDING_TYPES = ("SecurityFindingAnnotation", "VulnerabilityAnnotation", + "PenetrationTestFindingAnnotation") + SEVERITY_RANK = {"None": 0, "Low": 1, "Medium": 2, "High": 3, "Critical": 4} + + @staticmethod + def _report_of(annotation: dict) -> dict: + """Header and properties of the survey report an annotation is reported in, or {} if none is linked.""" + for rel in annotation.get("fromSurveyReports") or []: + related = rel.get("relatedElement") or {} + return {"guid": (related.get("elementHeader") or {}).get("guid"), **(related.get("properties") or {})} + return {} + + @classmethod + def _security_row(cls, annotation: dict) -> dict: + """One display row for a security finding or posture annotation. + + Egeria stores cvssScore, exploitAvailable, patchAvailable and the posture counts as Java + primitives, so an unset value comes back as 0 or false and cannot be told apart from a real one. + Where that would read as an answer nobody gave, the row leaves the value blank instead: + + - a CVSS score of 0.0 is only shown when the severity is "None", the rating a 0.0 score means; + - exploitAvailable and patchAvailable are only shown when true; + - a run with 0 resources in scope was not recorded, so both resource counts are blank. + + Counts by severity are read from findingCountsBySeverity: a severity missing from a recorded + map is 0, and with no map at all every severity count is blank. + """ + props = dict(annotation.get("properties") or {}) + type_name = ((annotation.get("elementHeader") or {}).get("type") or {}).get("typeName") + props["findingType"] = (type_name or "").removesuffix("Annotation") + + if props.get("cvssScore") in (0, 0.0) and props.get("severity") != "None": + props["cvssScore"] = None + for flag in ("exploitAvailable", "patchAvailable"): + if props.get(flag) is False: + props[flag] = None + + if type_name == "SecurityPostureAnnotation": + counts = props.get("findingCountsBySeverity") + for severity in cls.SEVERITY_RANK: + props[f"{severity.lower()}Findings"] = (counts.get(severity, 0) if isinstance(counts, dict) + else None) + if not props.get("resourcesInScopeCount"): + props["resourcesInScopeCount"] = None + if not props.get("resourcesTestedCount"): + props["resourcesTestedCount"] = None + + # Column keys are snake_case mapped to camelCase, which cannot produce packageURL. + if "packageURL" in props: + props["packageUrl"] = props["packageURL"] + + report = cls._report_of(annotation) + props["reportName"] = report.get("displayName") or report.get("qualifiedName") + props["reportCompletionTime"] = report.get("completionTime") + props["reportStartTime"] = report.get("startTime") + return {"elementHeader": annotation.get("elementHeader", {}), "properties": props} + + async def _async_resolve_guid(self, name_or_guid: str, type_name: str) -> str: + """`name_or_guid` itself if it is a GUID, otherwise the GUID of the element with that name.""" + try: + uuid.UUID(str(name_or_guid)) + return name_or_guid + except ValueError: + return await self._async_get_guid_for_name(name_or_guid, type_name=type_name) + + def _generate_security_output(self, elements: list[dict], output_format: str, + report_spec: str | dict, **kwargs) -> list | str: + rows = [self._security_row(e) for e in elements] + return self._generate_formatted_output( + elements=rows, + entity_type="Annotation", + output_format=output_format, + extract_properties_func=self._extract_annotation_properties, + report_spec=report_spec, + **kwargs, + ) + + @dynamic_catch + async def _async_get_security_findings( + self, + report_guid: Optional[str] = None, + element_guid: Optional[str] = None, + min_severity: Optional[str] = None, + finding_types: Optional[list[str]] = None, + output_format: str = "JSON", + report_spec: str | dict | None = None, + **kwargs, + ) -> list | str: + """Return the security findings in one survey report, or about one element, most severe first. Async version. + + Parameters + ---------- + report_guid : str, optional + The survey report (GUID or qualified name) whose findings to return. + element_guid : str, optional + The element (GUID or qualified name) whose findings to return, through AssociatedAnnotation. + Give exactly one of report_guid and element_guid. + min_severity : str, optional + Only findings at least this severe: None, Low, Medium, High or Critical. + finding_types : list[str], optional + Annotation types to include. Defaults to SecurityFindingAnnotation and its Vulnerability and + PenetrationTestFinding subtypes. + output_format : str, default "JSON" + JSON returns the annotations as Egeria returned them; other formats use the report spec. + report_spec : str | dict, optional + Defaults to "Security-Findings". + + Returns + ------- + list | str + The findings sorted by CVSS score, highest first (findings with no recorded score after + those with one, by severity), or NO_ELEMENTS_FOUND. + """ + if bool(report_guid) == bool(element_guid): + raise ValueError("Give exactly one of report_guid and element_guid.") + if min_severity is not None and min_severity not in self.SEVERITY_RANK: + raise ValueError(f"min_severity must be one of {', '.join(self.SEVERITY_RANK)}") + + if report_guid: + guid = await self._async_resolve_guid(report_guid, "SurveyReport") + annotations = await self._async_get_new_annotations(guid, output_format="JSON") + else: + guid = await self._async_resolve_guid(element_guid, None) + annotations = await self._async_get_annotations_for_element(guid, output_format="JSON") + if not isinstance(annotations, list): + return NO_ELEMENTS_FOUND + + wanted = set(finding_types or self.SECURITY_FINDING_TYPES) + floor = self.SEVERITY_RANK.get(min_severity, 0) + findings = [ + a for a in annotations + if ((a.get("elementHeader") or {}).get("type") or {}).get("typeName") in wanted + and (min_severity is None + or self.SEVERITY_RANK.get((a.get("properties") or {}).get("severity"), -1) >= floor) + ] + if not findings: + return NO_ELEMENTS_FOUND + + def sort_key(a: dict): + props = self._security_row(a)["properties"] + score = props.get("cvssScore") + return (score is None, -(score or 0), -self.SEVERITY_RANK.get(props.get("severity"), -1)) + + findings.sort(key=sort_key) + if output_format.upper() == "JSON": + return findings + return self._generate_security_output(findings, output_format, report_spec or "Security-Findings", **kwargs) + + @dynamic_catch + def get_security_findings( + self, + report_guid: Optional[str] = None, + element_guid: Optional[str] = None, + min_severity: Optional[str] = None, + finding_types: Optional[list[str]] = None, + output_format: str = "JSON", + report_spec: str | dict | None = None, + **kwargs, + ) -> list | str: + """Return the security findings in one survey report, or about one element, most severe first. + + See _async_get_security_findings. + """ + loop = asyncio.get_event_loop() + return loop.run_until_complete( + self._async_get_security_findings(report_guid, element_guid, min_severity, finding_types, + output_format, report_spec, **kwargs) + ) + + @dynamic_catch + async def _async_get_security_posture_history( + self, + report_qualified_name_prefix: Optional[str] = None, + output_format: str = "JSON", + report_spec: str | dict | None = None, + **kwargs, + ) -> list | str: + """Return the security posture summaries, one per test run, oldest first. Async version. + + Parameters + ---------- + report_qualified_name_prefix : str, optional + Only runs whose survey report's qualified name starts with this - for example + "SurveyReport::Supplier Portal::" for one system's series of tests. All runs when omitted. + output_format : str, default "JSON" + JSON returns the posture annotations as Egeria returned them; other formats use the report spec. + report_spec : str | dict, optional + Defaults to "Security-Posture". + + Returns + ------- + list | str + Sorted by the report's completion time (then start time, then the annotation's creation + time), or NO_ELEMENTS_FOUND. + """ + annotations = await self._async_find_annotations( + "*", metadata_element_type="SecurityPostureAnnotation", output_format="JSON", page_size=0) + if not isinstance(annotations, list): + return NO_ELEMENTS_FOUND + postures = [ + a for a in annotations + if ((a.get("elementHeader") or {}).get("type") or {}).get("typeName") == "SecurityPostureAnnotation" + and (not report_qualified_name_prefix + or str(self._report_of(a).get("qualifiedName") or "").startswith(report_qualified_name_prefix)) + ] + if not postures: + return NO_ELEMENTS_FOUND + + def run_time(a: dict) -> str: + report = self._report_of(a) + created = (((a.get("elementHeader") or {}).get("versions") or {}).get("createTime")) or "" + return str(report.get("completionTime") or report.get("startTime") or created) + + postures.sort(key=run_time) + if output_format.upper() == "JSON": + return postures + return self._generate_security_output(postures, output_format, report_spec or "Security-Posture", **kwargs) + + @dynamic_catch + def get_security_posture_history( + self, + report_qualified_name_prefix: Optional[str] = None, + output_format: str = "JSON", + report_spec: str | dict | None = None, + **kwargs, + ) -> list | str: + """Return the security posture summaries, one per test run, oldest first. + + See _async_get_security_posture_history. + """ + loop = asyncio.get_event_loop() + return loop.run_until_complete( + self._async_get_security_posture_history(report_qualified_name_prefix, output_format, + report_spec, **kwargs) + ) + # # Analysis Reports (deprecated) # diff --git a/pyegeria/view/base_report_formats.py b/pyegeria/view/base_report_formats.py index eb7d63e3..c4f1f42d 100644 --- a/pyegeria/view/base_report_formats.py +++ b/pyegeria/view/base_report_formats.py @@ -2241,6 +2241,107 @@ def _user_report_specs_dir() -> str: ) ), + "Security-Findings": FormatSet( + target_type="SecurityFindingAnnotation", + heading="Security Findings", + description="Security findings from one survey report, or about one element, most severe first. " + "A blank CVSS score, Exploit Available or Patch Available means not recorded: Egeria " + "returns 0 or false for these when they were never set.", + aliases=["Vulnerabilities", "Security Finding"], + family="Security Officer", + formats=[ + Format( + types=["ALL"], + attributes=[ + Column(name="Finding", key="display_name"), + Column(name="Type", key="finding_type"), + Column(name="Severity", key="severity"), + Column(name="CVSS Score", key="cvss_score"), + Column(name="Vulnerabilities", key="vulnerability_identifiers"), + Column(name="Weaknesses", key="weakness_identifiers"), + Column(name="Affected Location", key="affected_location"), + Column(name="Exploit Available", key="exploit_available"), + Column(name="Package URL", key="package_url"), + Column(name="Installed Version", key="installed_version"), + Column(name="Fixed Version", key="fixed_version"), + Column(name="Patch Available", key="patch_available"), + Column(name="Identifier", key="identifier"), + Column(name="Report", key="report_name"), + Column(name="GUID", key="GUID"), + ], + ), + ], + action=ActionParameter( + function="DataDiscovery.get_security_findings", + optional_params=["report_guid", "element_guid", "min_severity", "finding_types"], + spec_params={"output_format": "DICT"}, + ), + ), + "Penetration-Test-Findings": FormatSet( + target_type="PenetrationTestFindingAnnotation", + heading="Penetration Test Findings", + description="The findings of a penetration test, most severe first - the view the tester hands back. " + "A blank CVSS score means not recorded.", + aliases=["Pen Test Findings"], + family="Security Officer", + formats=[ + Format( + types=["ALL"], + attributes=[ + Column(name="Finding", key="display_name"), + Column(name="Severity", key="severity"), + Column(name="CVSS Score", key="cvss_score"), + Column(name="Attack Vector", key="attack_vector"), + Column(name="ATT&CK Techniques", key="attack_techniques"), + Column(name="Privileges Gained", key="privileges_gained"), + Column(name="Compromised Resources", key="compromised_resources"), + Column(name="Remediation", key="remediation"), + Column(name="Reproduction Steps", key="reproduction_steps"), + Column(name="Report", key="report_name"), + Column(name="GUID", key="GUID"), + ], + ), + ], + action=ActionParameter( + function="DataDiscovery.get_security_findings", + optional_params=["report_guid", "element_guid", "min_severity"], + spec_params={"output_format": "DICT", "finding_types": ["PenetrationTestFindingAnnotation"]}, + ), + ), + "Security-Posture": FormatSet( + target_type="SecurityPostureAnnotation", + heading="Security Posture", + description="One row per security test run, oldest first, to show whether the estate is getting more " + "secure. Blank severity counts mean the run recorded no counts; blank resource counts mean " + "they were not recorded. New and resolved counts of 0 may also be unrecorded - Egeria cannot " + "tell the two apart.", + aliases=["Security Posture History"], + family="Security Officer", + formats=[ + Format( + types=["ALL"], + attributes=[ + Column(name="Report", key="report_name"), + Column(name="Completed", key="report_completion_time"), + Column(name="Critical", key="critical_findings"), + Column(name="High", key="high_findings"), + Column(name="Medium", key="medium_findings"), + Column(name="Low", key="low_findings"), + Column(name="None", key="none_findings"), + Column(name="Resources In Scope", key="resources_in_scope_count"), + Column(name="Resources Tested", key="resources_tested_count"), + Column(name="New Findings", key="new_finding_count"), + Column(name="Resolved Findings", key="resolved_finding_count"), + Column(name="GUID", key="GUID"), + ], + ), + ], + action=ActionParameter( + function="DataDiscovery.get_security_posture_history", + optional_params=["report_qualified_name_prefix"], + spec_params={"output_format": "DICT"}, + ), + ), "Statement-of-Applicability-Controls": FormatSet( target_type="ControlApplicability", heading="Statement of Applicability Controls", diff --git a/tests/micro-tests/test_security_annotations.py b/tests/micro-tests/test_security_annotations.py new file mode 100644 index 00000000..9b972ff0 --- /dev/null +++ b/tests/micro-tests/test_security_annotations.py @@ -0,0 +1,179 @@ +""" +Security annotation models and the DataDiscovery security finding / posture views (Egeria 6.3, model 0665). +No live server: the annotation lists are stubbed, in the shape a live 6.3 platform returns. + +Egeria stores cvssScore, exploitAvailable, patchAvailable and the posture counts as Java primitives, so an +unset value comes back as 0 / false (confirmed live). The display rows must not present those as answers. +""" +import pytest +from pydantic import ValidationError + +from pyegeria.models import ( + PenetrationTestFindingAnnotationProperties, + SecurityPostureAnnotationProperties, + VulnerabilityAnnotationProperties, +) +from pyegeria.omvs.data_discovery import AnnotationProperties, DataDiscovery + + +def _annotation(type_name, guid, props, report=None): + element = { + "elementHeader": {"guid": guid, "type": {"typeName": type_name}, + "versions": {"createTime": f"2026-10-26T00:00:0{guid[-1]}Z"}}, + "properties": {"class": f"{type_name}Properties", "typeName": type_name, "displayName": guid, **props}, + } + if report: + element["fromSurveyReports"] = [{ + "relationshipHeader": {"type": {"typeName": "ReportedAnnotation"}}, + "relatedElement": {"elementHeader": {"guid": report["guid"], "type": {"typeName": "SurveyReport"}}, + "properties": {k: v for k, v in report.items() if k != "guid"}}, + }] + return element + + +# An unset vulnerability, as returned live: only severity was supplied. +SPARSE = _annotation("VulnerabilityAnnotation", "sparse-1", + {"severity": "Medium", "cvssScore": 0.0, "exploitAvailable": False, "patchAvailable": False}) +FULL = _annotation("VulnerabilityAnnotation", "full-2", + {"severity": "Critical", "cvssScore": 10.0, "exploitAvailable": True, + "packageURL": "pkg:maven/org.apache.logging.log4j/log4j-core@2.14.1"}) +PENTEST = _annotation("PenetrationTestFindingAnnotation", "pen-3", {"severity": "High", "cvssScore": 8.6}) +NONE_SCORE = _annotation("SecurityFindingAnnotation", "info-4", {"severity": "None", "cvssScore": 0.0}) +OTHER = _annotation("ResourceProfileAnnotation", "other-5", {}) + + +def test_models_serialize_egeria_field_names(): + v = VulnerabilityAnnotationProperties(**{"class": "VulnerabilityAnnotationProperties", "qualifiedName": "q", + "displayName": "d", "severity": "High", "cvssScore": 7.5, + "packageURL": "pkg:x", "patchAvailable": True}) + assert v.model_dump(by_alias=True, exclude_none=True) == { + "class": "VulnerabilityAnnotationProperties", "qualifiedName": "q", "displayName": "d", + "severity": "High", "cvssScore": 7.5, "packageURL": "pkg:x", "patchAvailable": True} + p = SecurityPostureAnnotationProperties(**{"class": "SecurityPostureAnnotationProperties", "qualifiedName": "q", + "displayName": "d", "findingCountsBySeverity": {"High": 4}, + "resourcesInScopeCount": 3}) + assert p.model_dump(by_alias=True, exclude_none=True)["findingCountsBySeverity"] == {"High": 4} + pen = PenetrationTestFindingAnnotationProperties(**{"class": "PenetrationTestFindingAnnotationProperties", + "qualifiedName": "q", "displayName": "d", + "attackTechniques": ["T1190"]}) + assert pen.model_dump(by_alias=True, exclude_none=True)["attackTechniques"] == ["T1190"] + + +def test_models_reject_bad_severity_and_score(): + with pytest.raises(ValidationError): + VulnerabilityAnnotationProperties(**{"class": "VulnerabilityAnnotationProperties", "qualifiedName": "q", + "displayName": "d", "severity": "Severe"}) + with pytest.raises(ValidationError): + VulnerabilityAnnotationProperties(**{"class": "VulnerabilityAnnotationProperties", "qualifiedName": "q", + "displayName": "d", "cvssScore": 11}) + + +def test_annotation_properties_uses_egerias_confidence_field(): + # The old local model used confidence_level, which Egeria silently ignores. + a = AnnotationProperties(**{"class": "AnnotationProperties", "qualifiedName": "q", "displayName": "d", + "confidence": 80}) + assert a.model_dump(by_alias=True, exclude_none=True)["confidence"] == 80 + + +def test_unrecorded_primitives_are_blank_not_zero_or_false(): + row = DataDiscovery._security_row(SPARSE)["properties"] + assert row["cvssScore"] is None + assert row["exploitAvailable"] is None + assert row["patchAvailable"] is None + assert row["findingType"] == "Vulnerability" + + +def test_real_values_are_kept(): + assert DataDiscovery._security_row(FULL)["properties"]["cvssScore"] == 10.0 + assert DataDiscovery._security_row(FULL)["properties"]["exploitAvailable"] is True + assert DataDiscovery._security_row(FULL)["properties"]["packageUrl"].startswith("pkg:maven") + # A 0.0 score is real when the severity says None. + assert DataDiscovery._security_row(NONE_SCORE)["properties"]["cvssScore"] == 0.0 + + +def test_posture_counts(): + report = {"guid": "r1", "displayName": "Run 1", "completionTime": "2026-10-01T00:00:00Z"} + recorded = _annotation("SecurityPostureAnnotation", "p-1", + {"findingCountsBySeverity": {"High": 1}, "resourcesInScopeCount": 3, + "resourcesTestedCount": 0, "newFindingCount": 1}, report) + row = DataDiscovery._security_row(recorded)["properties"] + assert (row["criticalFindings"], row["highFindings"], row["lowFindings"]) == (0, 1, 0) + assert row["resourcesInScopeCount"] == 3 + assert row["resourcesTestedCount"] == 0 # real: in scope was recorded + assert row["reportName"] == "Run 1" + + unrecorded = _annotation("SecurityPostureAnnotation", "p-2", + {"resourcesInScopeCount": 0, "resourcesTestedCount": 0}) + row = DataDiscovery._security_row(unrecorded)["properties"] + assert row["highFindings"] is None + assert row["resourcesInScopeCount"] is None and row["resourcesTestedCount"] is None + + +def _client(monkeypatch, annotations): + monkeypatch.setattr(DataDiscovery, "check_connection", lambda self: "test") + client = DataDiscovery("view-server", "https://localhost:1", "user", "pwd") + calls = [] + + async def new_annotations(guid, **kwargs): + calls.append(("report", guid)) + return annotations + + async def for_element(guid, **kwargs): + calls.append(("element", guid)) + return annotations + + async def find(search, **kwargs): + calls.append(("find", kwargs.get("metadata_element_type"))) + return annotations + + client._async_get_new_annotations = new_annotations + client._async_get_annotations_for_element = for_element + client._async_find_annotations = find + return client, calls + + +REPORT_GUID = "9696d446-58fb-4bc0-a473-7ece72212671" + + +@pytest.mark.asyncio +async def test_findings_sorted_by_cvss_with_unrecorded_last(monkeypatch): + client, calls = _client(monkeypatch, [SPARSE, PENTEST, OTHER, FULL]) + findings = await client._async_get_security_findings(report_guid=REPORT_GUID) + assert calls == [("report", REPORT_GUID)] + assert [f["elementHeader"]["guid"] for f in findings] == ["full-2", "pen-3", "sparse-1"] + + +@pytest.mark.asyncio +async def test_findings_filters(monkeypatch): + client, calls = _client(monkeypatch, [SPARSE, PENTEST, FULL]) + high = await client._async_get_security_findings(element_guid=REPORT_GUID, min_severity="High") + assert calls == [("element", REPORT_GUID)] + assert [f["elementHeader"]["guid"] for f in high] == ["full-2", "pen-3"] + pen = await client._async_get_security_findings(report_guid=REPORT_GUID, + finding_types=["PenetrationTestFindingAnnotation"]) + assert [f["elementHeader"]["guid"] for f in pen] == ["pen-3"] + + +@pytest.mark.asyncio +async def test_findings_argument_checks(monkeypatch): + client, _ = _client(monkeypatch, []) + with pytest.raises(ValueError): + await client._async_get_security_findings() + with pytest.raises(ValueError): + await client._async_get_security_findings(report_guid=REPORT_GUID, element_guid=REPORT_GUID) + with pytest.raises(ValueError): + await client._async_get_security_findings(report_guid=REPORT_GUID, min_severity="Severe") + + +@pytest.mark.asyncio +async def test_posture_history_oldest_first_and_prefix(monkeypatch): + later = _annotation("SecurityPostureAnnotation", "p-1", {}, + {"guid": "r2", "qualifiedName": "SurveyReport::Portal::Q2", "completionTime": "2027-01-16"}) + earlier = _annotation("SecurityPostureAnnotation", "p-2", {}, + {"guid": "r1", "qualifiedName": "SurveyReport::Portal::Q1", "completionTime": "2026-10-16"}) + other = _annotation("SecurityPostureAnnotation", "p-3", {}, + {"guid": "r3", "qualifiedName": "SurveyReport::Other::Q1", "completionTime": "2026-01-01"}) + client, calls = _client(monkeypatch, [later, other, earlier]) + history = await client._async_get_security_posture_history("SurveyReport::Portal::") + assert calls == [("find", "SecurityPostureAnnotation")] + assert [p["elementHeader"]["guid"] for p in history] == ["p-2", "p-1"]