diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index 09abb66b4..2439cf82f 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -1067,6 +1067,15 @@ def run(args: argparse.Namespace) -> None: # pragma: no cover from application.utils.external_project_parsers.parsers import dsomm BaseParser().register_resource(dsomm.DSOMM, db_connection_str=args.cache_file) + if getattr(args, "agent_evidence_vocabulary_in", False): + from application.utils.external_project_parsers.parsers import ( + agent_evidence_vocabulary, + ) + + BaseParser().register_resource( + agent_evidence_vocabulary.AgentEvidenceVocabulary, + db_connection_str=args.cache_file, + ) if args.cloud_native_security_controls_in: from application.utils.external_project_parsers.parsers import ( cloud_native_security_controls, diff --git a/application/tests/agent_evidence_vocabulary_cli_test.py b/application/tests/agent_evidence_vocabulary_cli_test.py new file mode 100644 index 000000000..aeb9c592e --- /dev/null +++ b/application/tests/agent_evidence_vocabulary_cli_test.py @@ -0,0 +1,38 @@ +from cre_logging import get_logger + +logger = get_logger(__name__) + +import argparse +import unittest +from typing import Any +from unittest.mock import patch + +from application.cmd import cre_main +from application.utils.external_project_parsers.base_parser import BaseParser +from application.utils.external_project_parsers.parsers import ( + agent_evidence_vocabulary, +) + + +class OnlyFlagsGiven(argparse.Namespace): + """Answer False for every flag the test does not set.""" + + def __getattr__(self, name: str) -> Any: + return False + + +class TestAgentEvidenceVocabularyCli(unittest.TestCase): + @patch.object(BaseParser, "register_resource") + def test_flag_registers_the_parser(self, register_resource) -> None: + cre_main.run( + OnlyFlagsGiven(agent_evidence_vocabulary_in=True, cache_file="cache.sqlite") + ) + register_resource.assert_called_once_with( + agent_evidence_vocabulary.AgentEvidenceVocabulary, + db_connection_str="cache.sqlite", + ) + + @patch.object(BaseParser, "register_resource") + def test_parser_does_not_run_without_the_flag(self, register_resource) -> None: + cre_main.run(OnlyFlagsGiven(cache_file="cache.sqlite")) + register_resource.assert_not_called() diff --git a/application/tests/agent_evidence_vocabulary_parser_test.py b/application/tests/agent_evidence_vocabulary_parser_test.py new file mode 100644 index 000000000..6fa92cf51 --- /dev/null +++ b/application/tests/agent_evidence_vocabulary_parser_test.py @@ -0,0 +1,373 @@ +from cre_logging import get_logger + +logger = get_logger(__name__) + +import unittest +from typing import Any, Dict, List, Optional +from unittest.mock import patch + +import requests + +from application import create_app, sqla # type: ignore +from application.database import db +from application.defs import cre_defs as defs +from application.prompt_client import prompt_client +from application.utils.external_project_parsers.parsers import ( + agent_evidence_vocabulary, +) + +TAGS_URL = "https://api.github.com/repos/probityai/agent-evidence/tags?per_page=100" +TAGS_URL_PAGE_2 = TAGS_URL + "&page=2" +RAW_PREFIX = "https://raw.githubusercontent.com/probityai/agent-evidence/" +BLOB_PREFIX = "https://github.com/probityai/agent-evidence/blob/" +VOCABULARY_PATH = "packages/vocabulary/vocabulary.yaml" + + +class FakeResponse: + def __init__( + self, + status_code: int = 200, + text: str = "", + json_body: Optional[List[Dict[str, Any]]] = None, + next_url: Optional[str] = None, + ) -> None: + self.status_code = status_code + self.text = text + self._json = json_body or [] + self.links = {"next": {"url": next_url}} if next_url else {} + + def json(self) -> List[Dict[str, Any]]: + return self._json + + +def tag_list(*names: str) -> List[Dict[str, Any]]: + return [{"name": n} for n in names] + + +def line_of(text: str, key: str) -> int: + for number, line in enumerate(text.splitlines(), start=1): + if line == f" {key}:": + return number + raise AssertionError(f"{key} not in fixture") + + +class TestAgentEvidenceVocabularyParser(unittest.TestCase): + def tearDown(self) -> None: + self.app_context.pop() + + def setUp(self) -> None: + self.app = create_app(mode="test") + self.app_context = self.app.app_context() + self.app_context.push() + sqla.create_all() + self.collection = db.Node_collection() + self.parser = agent_evidence_vocabulary.AgentEvidenceVocabulary() + # 515-021 is deliberately absent, so containment_posture cannot resolve. + for cre_id, name in [ + ("026-280", "Securely transfer logs (remotely)"), + ("433-442", "Verification"), + ("124-564", "Configuration Management"), + ("547-824", "AI model performance validation"), + ]: + self.collection.add_cre(cre=defs.CRE(id=cre_id, name=name)) + + def fake_get(self, responses: Dict[str, FakeResponse]): + def get(url: str, *args: Any, **kwargs: Any) -> FakeResponse: + if url not in responses: + raise AssertionError(f"unexpected request to {url}") + return responses[url] + + return get + + @patch.object(requests, "get") + def test_parse(self, mock_get) -> None: + tag = "agent-evidence-vocabulary-v0.6.0" + mock_get.side_effect = self.fake_get( + { + TAGS_URL: FakeResponse( + json_body=tag_list( + "agent-evidence-vocabulary-python-v0.6.0", + "agent-evidence-vocabulary-v0.5.0", + tag, + "agent-evidence-vocabulary-v0.3.0", + "unrelated-v9.9.9", + ) + ), + f"{RAW_PREFIX}{tag}/{VOCABULARY_PATH}": FakeResponse( + text=VOCABULARY_YAML + ), + } + ) + + result = self.parser.parse( + cache=self.collection, + ph=prompt_client.PromptHandler(database=self.collection), + ) + + self.assertEqual(list(result.results.keys()), [self.parser.name]) + nodes = {n.sectionID: n for n in result.results[self.parser.name]} + self.assertCountEqual( + nodes.keys(), + [ + "evidence_dimensions.observation_vantage", + "posture_and_coverage.coverage_denominator", + "change_admission.admission_gate", + ], + ) + + tags = [ + "family:standard", + "subtype:requirements_standard", + "audience:architect", + "maturity:draft", + "source:agent_evidence_vocabulary", + ] + expected = defs.Standard( + name="Agent Evidence Vocabulary", + section="observation_vantage", + sectionID="evidence_dimensions.observation_vantage", + version="0.6.0", + hyperlink=f"{BLOB_PREFIX}{tag}/{VOCABULARY_PATH}#L{line_of(VOCABULARY_YAML, 'observation_vantage')}", + description=( + "Where the evidence for a claim was obtained from.\n" + "CRE 026-280 (partial): The record is kept where the executed " + "party cannot alter it." + ), + tags=tags, + ) + vantage = nodes["evidence_dimensions.observation_vantage"] + got = vantage.todict() + links = got.pop("links") + self.assertEqual( + got, {k: v for k, v in expected.todict().items() if k != "links"} + ) + self.assertEqual([l["document"]["id"] for l in links], ["026-280"]) + self.assertEqual( + [l["ltype"] for l in links], [defs.LinkTypes.AutomaticallyLinkedTo.value] + ) + + gate = nodes["change_admission.admission_gate"] + self.assertEqual([l.document.id for l in gate.links], ["124-564", "547-824"]) + self.assertTrue( + all(l.ltype == defs.LinkTypes.AutomaticallyLinkedTo for l in gate.links) + ) + self.assertIn("CRE 124-564 (structural):", gate.description) + self.assertIn("CRE 547-824 (partial):", gate.description) + self.assertTrue( + gate.hyperlink.endswith(f"#L{line_of(VOCABULARY_YAML, 'admission_gate')}") + ) + + def test_latest_release_tag_follows_pages_and_skips_older_releases( + self, + ) -> None: + responses = { + TAGS_URL: FakeResponse( + json_body=tag_list( + "agent-evidence-vocabulary-v0.4.0", + "agent-evidence-vocabulary-v0.10.0", + ), + next_url=TAGS_URL_PAGE_2, + ), + TAGS_URL_PAGE_2: FakeResponse( + json_body=tag_list( + "agent-evidence-vocabulary-v0.9.1", + "agent-evidence-vocabulary-v1.0.0-rc1", + ) + ), + } + with patch.object(requests, "get", side_effect=self.fake_get(responses)): + self.assertEqual( + self.parser.latest_release_tag(), + "agent-evidence-vocabulary-v0.10.0", + ) + + def test_latest_release_tag_refuses_releases_without_links(self) -> None: + responses = { + TAGS_URL: FakeResponse( + json_body=tag_list( + "agent-evidence-vocabulary-v0.3.0", + "agent-evidence-vocabulary-python-v0.6.0", + ) + ) + } + with patch.object(requests, "get", side_effect=self.fake_get(responses)): + with self.assertRaises(RuntimeError): + self.parser.latest_release_tag() + + def test_failed_download_raises(self) -> None: + tag = "agent-evidence-vocabulary-v0.6.0" + responses = { + TAGS_URL: FakeResponse(json_body=tag_list(tag)), + f"{RAW_PREFIX}{tag}/{VOCABULARY_PATH}": FakeResponse(status_code=404), + } + with patch.object(requests, "get", side_effect=self.fake_get(responses)): + with self.assertRaises(RuntimeError): + self.parser.parse(cache=self.collection, ph=None) + + def test_version_comes_from_the_source(self) -> None: + tag = "agent-evidence-vocabulary-v0.6.0" + declared = VOCABULARY_YAML.replace('version: "0.6.0"', 'version: "0.6.1"') + nodes = self.parser.parse_vocabulary( + text=declared, tag=tag, cache=self.collection + ) + self.assertEqual({n.version for n in nodes}, {"0.6.1"}) + + no_meta = VOCABULARY_YAML.replace("meta:\n", "meta_removed:\n") + nodes = self.parser.parse_vocabulary( + text=no_meta, tag=tag, cache=self.collection + ) + self.assertEqual({n.version for n in nodes}, {"0.6.0"}) + + def test_missing_cre_is_skipped_without_breaking_the_import(self) -> None: + nodes = self.parser.parse_vocabulary( + text=VOCABULARY_YAML, + tag="agent-evidence-vocabulary-v0.6.0", + cache=self.collection, + ) + section_ids = [n.sectionID for n in nodes] + self.assertNotIn("posture_and_coverage.containment_posture", section_ids) + self.assertIn("posture_and_coverage.coverage_denominator", section_ids) + + # once the CRE exists, the same term imports + self.collection.add_cre( + cre=defs.CRE(id="515-021", name="Sandbox and isolate applications") + ) + nodes = self.parser.parse_vocabulary( + text=VOCABULARY_YAML, + tag="agent-evidence-vocabulary-v0.6.0", + cache=self.collection, + ) + self.assertIn( + "posture_and_coverage.containment_posture", + [n.sectionID for n in nodes], + ) + + def test_malformed_entries_are_skipped(self) -> None: + nodes = self.parser.parse_vocabulary( + text=VOCABULARY_YAML, + tag="agent-evidence-vocabulary-v0.6.0", + cache=self.collection, + ) + by_id = {n.sectionID: n for n in nodes} + # bad URL, unknown match value, non-mapping item and opencre not a list + for broken in [ + "record_lifecycle.revision_reason", + "record_lifecycle.wrong_match", + "record_lifecycle.not_a_mapping", + "record_lifecycle.not_a_list", + ]: + self.assertNotIn(broken, by_id) + # a term with no opencre key is not imported + self.assertNotIn("evidence_dimensions.observation_directness", by_id) + # a CRE listed twice is linked once, and the good link next to a broken one stays + denominator = by_id["posture_and_coverage.coverage_denominator"] + self.assertEqual([l.document.id for l in denominator.links], ["433-442"]) + self.assertEqual(denominator.description.count("CRE 433-442"), 1) + + def test_top_level_must_be_a_mapping(self) -> None: + with self.assertRaises(ValueError): + self.parser.parse_vocabulary( + text="- a\n- b\n", + tag="agent-evidence-vocabulary-v0.6.0", + cache=self.collection, + ) + + +VOCABULARY_YAML = """# agent-evidence-vocabulary test fixture +meta: + version: "0.6.0" + status: "draft - open for review" + +evidence_dimensions: + observation_vantage: + status: proposed + values: [substrate, artifact] + definition: > + Where the evidence for a claim + was obtained from. + opencre: + - cre: https://www.opencre.org/cre/026-280 + match: partial + basis: "The record is kept where the executed party cannot alter it." + + observation_directness: + status: proposed + values: [intercepted, reconstructed] + definition: > + Whether a claim rests on events captured as they occurred. + +posture_and_coverage: + containment_posture: + definition: > + The network isolation the artifact ran inside. + opencre: + - cre: https://www.opencre.org/cre/515-021 + match: structural + basis: "The term records the isolation the artifact ran inside." + + coverage_denominator: + definition: > + The population a result assessed. + opencre: + - cre: https://www.opencre.org/cre/433-442 + match: partial + basis: "A test result names the population it assessed." + - cre: https://www.opencre.org/cre/433-442 + match: partial + basis: "The same CRE listed a second time." + - cre: https://example.com/cre/433-442 + match: partial + basis: "Not an opencre.org URL." + +record_lifecycle: + revision_reason: + definition: > + Why a published record changed. + opencre: + - cre: https://www.opencre.org/cre/757-27 + match: partial + basis: "The id is too short." + + wrong_match: + definition: > + A term whose match value is not in the closed set. + opencre: + - cre: https://www.opencre.org/cre/433-442 + match: close-enough + basis: "Unknown match value." + + not_a_mapping: + definition: > + A term whose link item is a bare string. + opencre: + - https://www.opencre.org/cre/433-442 + + not_a_list: + definition: > + A term whose opencre field is a single mapping. + opencre: + cre: https://www.opencre.org/cre/433-442 + match: partial + basis: "Not a list." + +change_admission: + admission_gate: + values: [replay, heldout] + definition: > + The closed names of the checks an admission rests on. + opencre: + - cre: https://www.opencre.org/cre/124-564 + match: structural + basis: "The checks a change passes before it is admitted." + - cre: https://www.opencre.org/cre/547-824 + match: partial + basis: "Held-out runs validate the changed agent's behaviour." + +crosswalk_match_types: + exact: + definition: > + Same meaning, same shape. + +out_of_scope: + - identity +""" diff --git a/application/utils/external_project_parsers/parsers/agent_evidence_vocabulary.py b/application/utils/external_project_parsers/parsers/agent_evidence_vocabulary.py new file mode 100644 index 000000000..60aaab914 --- /dev/null +++ b/application/utils/external_project_parsers/parsers/agent_evidence_vocabulary.py @@ -0,0 +1,244 @@ +"""Import the agent-evidence-vocabulary term registry as a standard. + +The vocabulary keeps its CRE links in its own source: a term whose definition +states a requirement that a Common Requirement organizes carries an ``opencre`` +list, one item per link:: + + observation_vantage: + definition: ... + opencre: + - cre: https://www.opencre.org/cre/026-280 + match: partial + basis: "one line on what the term and the requirement share" + +This parser reads ``vocabulary.yaml`` at the newest release tag (tags are named +``agent-evidence-vocabulary-vX.Y.Z``), makes one node per linked term and links +it to each CRE it names. ``match`` is one of ``exact``, ``structural`` or +``partial``; OpenCRE links carry no strength, so the match and its basis are +written into the node description. A CRE id the cache does not know, or a +malformed link, is logged and skipped without stopping the import. +""" + +from cre_logging import get_logger + +logger = get_logger(__name__) + +import re +from typing import Any, Dict, List, Optional, Tuple + +import requests +import yaml + +from application.database import db +from application.defs import cre_defs as defs +from application.prompt_client import prompt_client as prompt_client +from application.utils.external_project_parsers import base_parser_defs +from application.utils.external_project_parsers.base_parser_defs import ( + ParserInterface, + ParseResult, +) + + +class AgentEvidenceVocabulary(ParserInterface): + name = "Agent Evidence Vocabulary" + repository = "probityai/agent-evidence" + vocabulary_path = "packages/vocabulary/vocabulary.yaml" + tag_prefix = "agent-evidence-vocabulary-v" + # The first release whose terms carry opencre links. + minimum_version = "0.4.0" + match_values = ("exact", "structural", "partial") + cre_url = re.compile(r"^https://(www\.)?opencre\.org/cre/(?P\d{3}-\d{3})$") + tag_pattern = re.compile(r"^agent-evidence-vocabulary-v(?P\d+\.\d+\.\d+)$") + + def tags_url(self) -> str: + return f"https://api.github.com/repos/{self.repository}/tags?per_page=100" + + def raw_url(self, tag: str) -> str: + return f"https://raw.githubusercontent.com/{self.repository}/{tag}/{self.vocabulary_path}" + + def term_url(self, tag: str, line: int) -> str: + return f"https://github.com/{self.repository}/blob/{tag}/{self.vocabulary_path}#L{line}" + + def parse( + self, + cache: db.Node_collection, + ph: Optional[prompt_client.PromptHandler], + ) -> ParseResult: + tag = self.latest_release_tag() + resp = requests.get(self.raw_url(tag), timeout=60) + if resp.status_code != 200: + err_str = f"could not retrieve {self.vocabulary_path} at {tag}, status code {resp.status_code}" + logger.fatal(err_str) + raise RuntimeError(err_str) + entries = self.parse_vocabulary(text=resp.text, tag=tag, cache=cache) + results: Dict[str, List[defs.Document]] = {self.name: list(entries)} + base_parser_defs.validate_classification_tags(results) + return ParseResult(results=results) + + def latest_release_tag(self) -> str: + """Return the newest release tag at or above ``minimum_version``.""" + url: Optional[str] = self.tags_url() + releases: List[Tuple[Tuple[int, ...], str]] = [] + minimum = self.version_key(self.minimum_version) + while url: + resp = requests.get(url, timeout=60) + if resp.status_code != 200: + err_str = f"could not list tags of {self.repository}, status code {resp.status_code}" + logger.fatal(err_str) + raise RuntimeError(err_str) + for item in resp.json(): + name = item.get("name", "") if isinstance(item, dict) else "" + found = self.tag_pattern.match(name) + if not found: + continue + version = self.version_key(found.group("version")) + if version >= minimum: + releases.append((version, name)) + url = resp.links.get("next", {}).get("url") + if not releases: + err_str = f"no {self.tag_prefix}X.Y.Z tag at or above {self.minimum_version} in {self.repository}" + logger.fatal(err_str) + raise RuntimeError(err_str) + return max(releases)[1] + + @staticmethod + def version_key(version: str) -> Tuple[int, ...]: + return tuple(int(part) for part in version.split(".")) + + def term_lines(self, text: str) -> Dict[Tuple[str, str], int]: + """Map (group, term) to the 1-based line where the term's key starts.""" + lines: Dict[Tuple[str, str], int] = {} + root = yaml.compose(text, Loader=yaml.SafeLoader) + if not isinstance(root, yaml.MappingNode): + return lines + for group_key, group_node in root.value: + if not isinstance(group_node, yaml.MappingNode): + continue + for term_key, _ in group_node.value: + lines[(group_key.value, term_key.value)] = term_key.start_mark.line + 1 + return lines + + def parse_vocabulary( + self, text: str, tag: str, cache: db.Node_collection + ) -> List[defs.Standard]: + vocabulary = yaml.safe_load(text) + if not isinstance(vocabulary, dict): + raise ValueError(f"{self.vocabulary_path} at {tag} is not a mapping") + version = self.version_of(vocabulary, tag) + lines = self.term_lines(text) + entries: List[defs.Standard] = [] + for group, terms in vocabulary.items(): + if group == "meta" or not isinstance(terms, dict): + continue + for term, body in terms.items(): + if not isinstance(body, dict) or "opencre" not in body: + continue + standard = self.make_standard( + group=group, + term=term, + body=body, + version=version, + hyperlink=self.term_url(tag, lines.get((group, term), 1)), + cache=cache, + ) + if standard is not None: + entries.append(standard) + return entries + + def version_of(self, vocabulary: Dict[str, Any], tag: str) -> str: + tag_version = tag[len(self.tag_prefix) :] + meta = vocabulary.get("meta") + declared = meta.get("version") if isinstance(meta, dict) else None + if not declared: + return tag_version + if str(declared) != tag_version: + logger.warning( + f"{self.vocabulary_path} at {tag} declares version {declared}, using it" + ) + return str(declared) + + def make_standard( + self, + group: str, + term: str, + body: Dict[str, Any], + version: str, + hyperlink: str, + cache: db.Node_collection, + ) -> Optional[defs.Standard]: + links = body.get("opencre") + if not isinstance(links, list): + logger.error(f"{group}.{term}: opencre is not a list, skipping the term") + return None + definition = " ".join(str(body.get("definition") or "").split()) + standard = defs.Standard( + name=self.name, + section=term, + sectionID=f"{group}.{term}", + hyperlink=hyperlink, + version=version, + tags=base_parser_defs.build_tags( + family=base_parser_defs.Family.STANDARD, + subtype=base_parser_defs.Subtype.REQUIREMENTS_STANDARD, + audience=base_parser_defs.Audience.ARCHITECT, + maturity=base_parser_defs.Maturity.DRAFT, + source="agent_evidence_vocabulary", + extra=[], + ), + ) + link_notes: List[str] = [] + linked: List[str] = [] + for item in links: + parsed = self.parse_link(group, term, item) + if parsed is None: + continue + cre_id, match, basis = parsed + cres = cache.get_CREs(external_id=cre_id) + if not cres: + logger.warning( + f"{group}.{term}: CRE {cre_id} is not in the cache, skipping the link" + ) + continue + added = False + for cre in cres: + if cre.id in linked: + logger.warning( + f"{group}.{term}: CRE {cre_id} is listed twice, keeping the first" + ) + continue + standard.add_link( + defs.Link( + document=cre.shallow_copy(), + ltype=defs.LinkTypes.AutomaticallyLinkedTo, + ) + ) + linked.append(cre.id) + added = True + if added: + link_notes.append(f"CRE {cre_id} ({match}): {basis}") + if not standard.links: + logger.warning(f"{group}.{term}: no CRE link resolved, skipping the term") + return None + standard.description = "\n".join([definition] + link_notes).strip() + return standard + + def parse_link( + self, group: str, term: str, item: Any + ) -> Optional[Tuple[str, str, str]]: + if not isinstance(item, dict): + logger.error(f"{group}.{term}: opencre item {item!r} is not a mapping") + return None + found = self.cre_url.match(str(item.get("cre", ""))) + if not found: + logger.error( + f"{group}.{term}: {item.get('cre')!r} is not an opencre.org/cre/NNN-NNN URL" + ) + return None + match = item.get("match") + if match not in self.match_values: + logger.error( + f"{group}.{term}: match {match!r} is not one of {', '.join(self.match_values)}" + ) + return None + basis = " ".join(str(item.get("basis") or "").split()) + return found.group("cre"), match, basis diff --git a/cre.py b/cre.py index d420bb48c..16efcf1c3 100644 --- a/cre.py +++ b/cre.py @@ -222,6 +222,11 @@ def main() -> None: action="store_true", help="import dsomm from their repo (https://raw.githubusercontent.com/devsecopsmaturitymodel/DevSecOps-MaturityModel-data/main/src/assets/YAML/generated/generated.yaml)", ) + parser.add_argument( + "--agent_evidence_vocabulary_in", + action="store_true", + help="import the agent-evidence-vocabulary terms and their CRE links from vocabulary.yaml at its newest release tag (https://github.com/probityai/agent-evidence/tree/main/packages/vocabulary)", + ) parser.add_argument( "--cloud_native_security_controls_in", action="store_true",