From 45bee842ddb6bbe007b34bc14f744ed9fb650d2d Mon Sep 17 00:00:00 2001 From: Aryan Pardeshi Date: Tue, 27 Oct 2026 00:44:57 +0530 Subject: [PATCH] fix(oscal): export CRE lookups as valid OSCAL catalogs format=oscal returned 500 for /id, /name, root_cres and CRE text_search. document_to_oscal now defaults uuid/last_modified, uses a timezone-aware timestamp, skips unsupported links and omits empty controls. list_to_oscal builds controls for lists of CREs. Fixes #1176 --- application/tests/oscal_utils_test.py | 79 +++++++++++++++++++++++++++ application/tests/web_main_test.py | 65 ++++++++++++++++++++++ application/utils/oscal_utils.py | 32 ++++++++--- 3 files changed, 167 insertions(+), 9 deletions(-) diff --git a/application/tests/oscal_utils_test.py b/application/tests/oscal_utils_test.py index fc5040d4a..f98fb4006 100644 --- a/application/tests/oscal_utils_test.py +++ b/application/tests/oscal_utils_test.py @@ -3,6 +3,8 @@ logger = get_logger(__name__) import json +from datetime import datetime +from uuid import UUID from typing import Any, Dict, List, Tuple import unittest @@ -459,6 +461,83 @@ def test_tool_document_to_oscal(self) -> None: ) self.assertDictEqual(remove_empty_elements(result), expected) + def test_cre_document_to_oscal_without_uuid_and_last_modified(self) -> None: + cre = defs.CRE(name="cre-1", id="001-001", description="cre-desc") + cre.add_link( + defs.Link( + document=defs.Standard( + name="standard-1", + section="s1", + hyperlink="https://example.com/1", + ), + ltype=defs.LinkTypes.LinkedTo, + ) + ) + + result = json.loads(oscal_utils.document_to_oscal(cre)) + + self.assertEqual(str(UUID(result["uuid"])), result["uuid"]) + last_modified = datetime.fromisoformat(result["metadata"]["last_modified"]) + self.assertIsNotNone(last_modified.tzinfo) + self.assertEqual( + [c["id"] for c in result["controls"]], + ["_s1"], + ) + + def test_cre_document_to_oscal_without_links(self) -> None: + cre = defs.CRE(name="cre-1", id="001-001", description="cre-desc") + + result = json.loads(oscal_utils.document_to_oscal(cre)) + + self.assertEqual(result["metadata"]["title"], "cre-1") + self.assertFalse(result.get("controls")) + + def test_cre_document_to_oscal_skips_unsupported_link_types(self) -> None: + cre = defs.CRE(name="cre-1", id="001-001") + cre.add_link( + defs.Link( + document=defs.Code(name="code-1", hyperlink="https://example.com/c"), + ltype=defs.LinkTypes.LinkedTo, + ) + ) + cre.add_link( + defs.Link( + document=defs.CRE(name="cre-2", id="002-002"), + ltype=defs.LinkTypes.LinkedTo, + ) + ) + + result = json.loads(oscal_utils.document_to_oscal(cre)) + + self.assertEqual([c["id"] for c in result["controls"]], ["_002-002"]) + + def test_list_of_cres_to_oscal(self) -> None: + cres = [ + defs.CRE(name=f"cre-{i}", id=f"{i}{i}{i}-{i}{i}{i}", description="d") + for i in range(0, 3) + ] + + result = json.loads(oscal_utils.list_to_oscal(cres)) # type: ignore[arg-type] + + self.assertEqual(result["metadata"]["title"], "cre-0") + self.assertIsNotNone( + datetime.fromisoformat(result["metadata"]["last_modified"]).tzinfo + ) + self.assertEqual( + [ + (c["id"], c["title"], [link["href"] for link in c["links"]]) + for c in result["controls"] + ], + [ + ( + f"_{i}{i}{i}-{i}{i}{i}", + f"cre-{i}", + [f"https://opencre.org/cre/{i}{i}{i}-{i}{i}{i}"], + ) + for i in range(0, 3) + ], + ) + if __name__ == "__main__": unittest.main() diff --git a/application/tests/web_main_test.py b/application/tests/web_main_test.py index 13b30f921..e0c175bfc 100644 --- a/application/tests/web_main_test.py +++ b/application/tests/web_main_test.py @@ -541,6 +541,71 @@ def test_find_root_cres(self) -> None: self.assertEqual(json.loads(response.data.decode()), expected) self.assertEqual(200, response.status_code) + def _add_cre_linked_to_standards(self) -> defs.CRE: + collection = self.collection + cre = defs.CRE(id="111-111", name="Root CRE", description="root") + dbcre = collection.add_cre(cre) + for section in ("V1", "V2"): + node = collection.add_node( + defs.Standard( + name="ASVS", + section=section, + sectionID=section, + hyperlink=f"https://example.com/{section}", + ) + ) + collection.add_link(dbcre, node, defs.LinkTypes.LinkedTo) + return cre + + def test_cre_endpoints_oscal_format(self) -> None: + cre = self._add_cre_linked_to_standards() + + with self.app.test_client() as client: + # single CRE endpoints wrap the catalog in "data" + for path in ( + f"/rest/v1/id/{cre.id}?format=oscal", + "/rest/v1/name/Root%20CRE?format=oscal", + ): + response = client.get(path) + self.assertEqual(200, response.status_code, path) + catalog = json.loads(response.data.decode())["data"] + self.assertEqual("Root CRE", catalog["metadata"]["title"], path) + self.assertEqual( + ["_V1", "_V2"], + sorted(c["id"] for c in catalog["controls"]), + path, + ) + + # list endpoints return the catalog directly + for path in ( + "/rest/v1/root_cres?format=oscal", + f"/rest/v1/text_search?text=CRE:{cre.id}&format=oscal", + ): + response = client.get(path) + self.assertEqual(200, response.status_code, path) + catalog = json.loads(response.data.decode()) + self.assertEqual("Root CRE", catalog["metadata"]["title"], path) + self.assertEqual( + [{"id": "_111-111", "title": "Root CRE"}], + [{"id": c["id"], "title": c["title"]} for c in catalog["controls"]], + path, + ) + self.assertEqual( + ["https://opencre.org/cre/111-111"], + [link["href"] for link in catalog["controls"][0]["links"]], + path, + ) + + def test_cre_without_links_oscal_format(self) -> None: + cre = defs.CRE(id="222-222", name="Lonely CRE", description="no links") + self.collection.add_cre(cre) + + with self.app.test_client() as client: + response = client.get(f"/rest/v1/id/{cre.id}?format=oscal") + self.assertEqual(200, response.status_code) + catalog = json.loads(response.data.decode())["data"] + self.assertEqual("Lonely CRE", catalog["metadata"]["title"]) + def test_smartlink(self) -> None: self.maxDiff = None collection = db.Node_collection().with_graph() diff --git a/application/utils/oscal_utils.py b/application/utils/oscal_utils.py index 3f18a3073..b270f3f60 100644 --- a/application/utils/oscal_utils.py +++ b/application/utils/oscal_utils.py @@ -5,14 +5,16 @@ import random from trestle.oscal import catalog from trestle.oscal import common -from datetime import datetime +from datetime import datetime, timezone from application.defs import cre_defs as defs from uuid import uuid4 from typing import List, Optional, Union def document_to_oscal( - document: defs.Document, uuid: Optional[str], last_modified: Optional[str] + document: defs.Document, + uuid: Optional[str] = None, + last_modified: Optional[Union[str, datetime]] = None, ) -> str: """ document_to_oscal takes a single document @@ -32,7 +34,7 @@ def document_to_oscal( hyperlink = f"https://opencre.org/cre/{document.id}" if not last_modified: - last_modified = datetime.utcnow() + last_modified = datetime.now(timezone.utc) m = None if document.description: m = common.Metadata( @@ -76,15 +78,17 @@ def document_to_oscal( title=link.document.name, links=[common.Link(href=link.document.hyperlink)], ) - controls.append(ctrl) - c.controls = controls + if ctrl is not None: + controls.append(ctrl) + if controls: + c.controls = controls return c.json() def list_to_oscal(documents: List[defs.Standard | defs.Tool]) -> str: """ - list_to_oscal takes a list of standards or tooling rules - (they all need to be the same type and the same name) + list_to_oscal takes a list of standards, tooling rules or CREs + (they all need to be the same type; standards and tools the same name) and creates a set of controls from them """ version = "0.0" @@ -98,7 +102,16 @@ def list_to_oscal(documents: List[defs.Standard | defs.Tool]) -> str: c = catalog.Catalog(metadata=m, uuid=str(uuid4())) controls: List[catalog.Control] = [] - if documents[0].doctype == defs.Credoctypes.Standard: + if documents[0].doctype == defs.Credoctypes.CRE: + for doc in documents: + controls.append( + catalog.Control( + id=f"_{doc.id}", + title=doc.name, + links=[common.Link(href=f"https://opencre.org/cre/{doc.id}")], + ) + ) + elif documents[0].doctype == defs.Credoctypes.Standard: for doc in documents: props = [] @@ -133,5 +146,6 @@ def list_to_oscal(documents: List[defs.Standard | defs.Tool]) -> str: links=[common.Link(href=doc.hyperlink)], ) ) - c.controls = controls + if controls: + c.controls = controls return c.json()