Add AWS Config compliance-results-to-OSCAL-AR task

Adds a new 'aws-config-result-to-oscal-ar' task + transformer (mirroring the
existing tanium/osco task+transformer shape) that converts AWS Config
compliance evaluation results into OSCAL Assessment Results.

Input: the json shape returned by AWS Config's
get-compliance-details-by-config-rule / get-compliance-details-by-resource
APIs (a top-level EvaluationResults list of EvaluationResult objects). Field
names verified against:
  https://docs.aws.amazon.com/config/latest/APIReference/API_EvaluationResult.html
  https://docs.aws.amazon.com/config/latest/APIReference/API_EvaluationResultQualifier.html

Output: one OSCAL Result per input file, with one Observation per
EvaluationResult (method TEST-AUTOMATED), an inventory-item per distinct
AWS resource (deduplicated across repeated evaluations of the same
resource), and compliance-type/config-rule-name/etc. carried as props.
Control mapping is intentionally include-all (AWS Config rules aren't
mapped to specific catalog controls out of the box), same approach OSCO
takes for the analogous case.

Verified against a real installed trestle (editable install, Python 3.12):
- 11 transformer-level tests (tests/trestle/transforms/implementations/
  aws_config_test.py) against a schema-verified fixture -- inventory
  dedup, subject linkage, compliance-type/annotation props, missing
  optional fields, full oscal_serialize_json_bytes round-trip.
- 6 task-level tests (tests/trestle/tasks/aws_config_result_to_oscal_ar_test.py)
  exercising the real TaskBase.execute()/simulate() path with actual file
  I/O, output-overwrite handling, and missing-config failure modes.
- Confirmed 'trestle task -l' lists the new task correctly alongside
  osco-result-to-oscal-ar and tanium-result-to-oscal-ar.
- flake8 clean on both new source files.

Two real bugs were caught and fixed by this testing before submission:
InventoryItem does not have a 'status' field (that's a SystemComponent
field; my first draft wrongly copied it), and OSCAL's Observation.props
requires min_length=1 when present, so it must be omitted rather than
passed as an empty list.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: AAH20

tests/data/tasks/aws-config/aws-config-sample.json

@@ -0,0 +1,67 @@
+{
+  "EvaluationResults": [
+    {
+      "EvaluationResultIdentifier": {
+        "EvaluationResultQualifier": {
+          "ConfigRuleName": "s3-bucket-public-read-prohibited",
+          "ResourceType": "AWS::S3::Bucket",
+          "ResourceId": "acme-prod-uploads",
+          "EvaluationMode": "DETECTIVE"
+        },
+        "OrderingTimestamp": "2026-08-16T09:12:03.000Z"
+      },
+      "ComplianceType": "NON_COMPLIANT",
+      "ResultRecordedTime": "2026-08-16T09:12:05.000Z",
+      "ConfigRuleInvokedTime": "2026-08-16T09:12:04.000Z",
+      "Annotation": "The S3 bucket policy allows public read access.",
+      "ResultToken": "tok-0001"
+    },
+    {
+      "EvaluationResultIdentifier": {
+        "EvaluationResultQualifier": {
+          "ConfigRuleName": "s3-bucket-public-read-prohibited",
+          "ResourceType": "AWS::S3::Bucket",
+          "ResourceId": "acme-prod-assets",
+          "EvaluationMode": "DETECTIVE"
+        },
+        "OrderingTimestamp": "2026-08-16T09:12:07.000Z"
+      },
+      "ComplianceType": "COMPLIANT",
+      "ResultRecordedTime": "2026-08-16T09:12:08.000Z",
+      "ConfigRuleInvokedTime": "2026-08-16T09:12:07.000Z",
+      "ResultToken": "tok-0002"
+    },
+    {
+      "EvaluationResultIdentifier": {
+        "EvaluationResultQualifier": {
+          "ConfigRuleName": "restricted-ssh",
+          "ResourceType": "AWS::EC2::SecurityGroup",
+          "ResourceId": "sg-0a1b2c3d4e5f6g7h8",
+          "EvaluationMode": "DETECTIVE"
+        },
+        "OrderingTimestamp": "2026-08-16T09:13:11.000Z"
+      },
+      "ComplianceType": "NON_COMPLIANT",
+      "ResultRecordedTime": "2026-08-16T09:13:12.000Z",
+      "ConfigRuleInvokedTime": "2026-08-16T09:13:11.000Z",
+      "Annotation": "Security group allows unrestricted SSH access (0.0.0.0/0 on port 22).",
+      "ResultToken": "tok-0003"
+    },
+    {
+      "EvaluationResultIdentifier": {
+        "EvaluationResultQualifier": {
+          "ConfigRuleName": "restricted-ssh",
+          "ResourceType": "AWS::EC2::SecurityGroup",
+          "ResourceId": "sg-0a1b2c3d4e5f6g7h8",
+          "EvaluationMode": "DETECTIVE"
+        },
+        "OrderingTimestamp": "2026-08-15T09:13:11.000Z"
+      },
+      "ComplianceType": "NON_COMPLIANT",
+      "ResultRecordedTime": "2026-08-15T09:13:12.000Z",
+      "ConfigRuleInvokedTime": "2026-08-15T09:13:11.000Z",
+      "Annotation": "Security group allows unrestricted SSH access (0.0.0.0/0 on port 22).",
+      "ResultToken": "tok-0000"
+    }
+  ]
+}

tests/trestle/tasks/aws_config_result_to_oscal_ar_test.py

@@ -0,0 +1,85 @@
+# -*- mode:python; coding:utf-8 -*-
+# Copyright (c) 2026 IBM Corp. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""AWS Config to OSCAL task tests."""
+
+import configparser
+import json
+import pathlib
+
+from tests.test_utils import TEST_DIR
+
+from trestle.tasks.aws_config_result_to_oscal_ar import AwsConfigResultToOscalAR
+from trestle.tasks.base_task import TaskOutcome
+
+test_data_dir = TEST_DIR / 'data/tasks/aws-config'
+
+
+def _build_config(input_dir: pathlib.Path, output_dir: pathlib.Path) -> configparser.SectionProxy:
+    config = configparser.ConfigParser()
+    section = 'task.aws-config-result-to-oscal-ar'
+    config.add_section(section)
+    config.set(section, 'input-dir', str(input_dir))
+    config.set(section, 'output-dir', str(output_dir))
+    return config[section]
+
+
+class TestAwsConfigResultToOscalAR:
+    def test_print_info(self, capsys):
+        task = AwsConfigResultToOscalAR(None)
+        task.print_info()
+
+    def test_missing_config_fails(self):
+        task = AwsConfigResultToOscalAR(None)
+        assert task.execute() == TaskOutcome.FAILURE
+
+    def test_missing_required_keys_fails(self):
+        config = configparser.ConfigParser()
+        config.add_section('task.aws-config-result-to-oscal-ar')
+        task = AwsConfigResultToOscalAR(config['task.aws-config-result-to-oscal-ar'])
+        assert task.execute() == TaskOutcome.FAILURE
+
+    def test_simulate_does_not_write_output(self, tmp_path):
+        output_dir = tmp_path / 'out'
+        config = _build_config(test_data_dir, output_dir)
+        task = AwsConfigResultToOscalAR(config)
+        outcome = task.simulate()
+        assert outcome == TaskOutcome.SIM_SUCCESS
+        assert not output_dir.exists() or not list(output_dir.iterdir())
+
+    def test_execute_produces_valid_oscal_result(self, tmp_path):
+        output_dir = tmp_path / 'out'
+        config = _build_config(test_data_dir, output_dir)
+        task = AwsConfigResultToOscalAR(config)
+        outcome = task.execute()
+        assert outcome == TaskOutcome.SUCCESS
+
+        produced = list(output_dir.glob('*.oscal.json'))
+        assert len(produced) == 1
+        data = json.loads(produced[0].read_text(encoding='utf-8'))
+        assert 'results' in data
+        result = data['results'][0]
+        assert len(result['observations']) == 4
+        assert len(result['local-definitions']['inventory-items']) == 3
+
+    def test_execute_respects_output_overwrite_false(self, tmp_path):
+        output_dir = tmp_path / 'out'
+        config = _build_config(test_data_dir, output_dir)
+        task = AwsConfigResultToOscalAR(config)
+        assert task.execute() == TaskOutcome.SUCCESS
+
+        config2 = _build_config(test_data_dir, output_dir)
+        config2['output-overwrite'] = 'false'
+        task2 = AwsConfigResultToOscalAR(config2)
+        assert task2.execute() == TaskOutcome.FAILURE

tests/trestle/transforms/implementations/aws_config_test.py

@@ -0,0 +1,144 @@
+# -*- mode:python; coding:utf-8 -*-
+# Copyright (c) 2026 IBM Corp. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for AwsConfigResultToOscalARTransformer.
+
+The fixture (tests/data/tasks/aws-config/aws-config-sample.json) is
+hand-constructed to match AWS Config's documented EvaluationResult schema
+field-for-field (EvaluationResultIdentifier / EvaluationResultQualifier /
+ComplianceType / ResultRecordedTime / ConfigRuleInvokedTime / Annotation /
+ResultToken), verified against
+https://docs.aws.amazon.com/config/latest/APIReference/API_EvaluationResult.html
+and API_EvaluationResultQualifier.html -- it is not a captured real AWS API
+response (no AWS account was used), but every field name and shape is real.
+"""
+import pathlib
+
+import pytest
+
+from trestle.transforms.implementations.aws_config import AwsConfigResultToOscalARTransformer
+
+test_data_dir = pathlib.Path('tests/data/tasks/aws-config').resolve()
+
+
+@pytest.fixture
+def sample_blob() -> str:
+    return (test_data_dir / 'aws-config-sample.json').read_text(encoding='utf-8')
+
+
+class TestTransform:
+    def test_produces_one_result_with_all_observations(self, sample_blob):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        assert len(results.root) == 1
+        result = results.root[0]
+        # 4 EvaluationResults in the fixture -> 4 Observations (one per evaluation, even
+        # when the same resource is evaluated more than once).
+        assert len(result.observations) == 4
+
+    def test_inventory_deduplicated_by_resource(self, sample_blob):
+        """The same security group appears in 2 EvaluationResults -> 1 inventory item."""
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        result = results.root[0]
+        inventory_items = result.local_definitions.inventory_items
+        # 3 distinct resources: 2 S3 buckets + 1 security group (evaluated twice).
+        assert len(inventory_items) == 3
+        descriptions = {i.description for i in inventory_items}
+        assert descriptions == {
+            'AWS::S3::Bucket acme-prod-uploads',
+            'AWS::S3::Bucket acme-prod-assets',
+            'AWS::EC2::SecurityGroup sg-0a1b2c3d4e5f6g7h8',
+        }
+
+    def test_observation_subjects_reference_the_correct_inventory_item(self, sample_blob):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        result = results.root[0]
+        sg_item_uuid = next(
+            i.uuid for i in result.local_definitions.inventory_items
+            if 'sg-0a1b2c3d4e5f6g7h8' in i.description
+        )
+        sg_observations = [o for o in result.observations if 'restricted-ssh' in o.description]
+        assert len(sg_observations) == 2
+        for obs in sg_observations:
+            assert obs.subjects[0].subject_uuid == sg_item_uuid
+            assert obs.subjects[0].type.root == 'inventory-item'
+
+    def test_compliance_type_and_annotation_carried_in_props(self, sample_blob):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        result = results.root[0]
+        non_compliant = [
+            o for o in result.observations
+            if any(p.name == 'compliance-type' and p.value == 'NON_COMPLIANT' for p in o.props)
+        ]
+        # 3 of the 4 fixture entries are NON_COMPLIANT.
+        assert len(non_compliant) == 3
+        s3_obs = next(o for o in result.observations if 'acme-prod-uploads' in o.description)
+        assert 'public read access' in s3_obs.description
+        assert 'AWS::S3::Bucket' in s3_obs.description
+
+    def test_compliant_entry_has_no_annotation_but_valid_description(self, sample_blob):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        result = results.root[0]
+        compliant_obs = next(o for o in result.observations if 'acme-prod-assets' in o.description)
+        assert compliant_obs.description == 's3-bucket-public-read-prohibited (AWS::S3::Bucket acme-prod-assets)'
+
+    def test_methods_and_collected_are_set(self, sample_blob):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        result = results.root[0]
+        for obs in result.observations:
+            assert obs.methods == ['TEST-AUTOMATED']
+            assert obs.collected is not None
+
+    def test_reviewed_controls_present(self, sample_blob):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        result = results.root[0]
+        assert result.reviewed_controls is not None
+        assert result.reviewed_controls.control_selections is not None
+
+    def test_analysis_reports_correct_counts(self, sample_blob):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        assert results.root  # ensure transform ran before checking analysis
+        assert 'inventory: 3' in transformer.analysis
+        assert 'observations: 4' in transformer.analysis
+
+    def test_empty_evaluation_results_produces_result_with_no_observations(self):
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform('{"EvaluationResults": []}')
+        assert len(results.root) == 1
+        assert results.root[0].observations is None
+
+    def test_missing_optional_fields_do_not_raise(self):
+        """A minimal, spec-legal EvaluationResult (all fields optional per AWS docs)."""
+        blob = '{"EvaluationResults": [{}]}'
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(blob)
+        result = results.root[0]
+        assert len(result.observations) == 1
+        assert result.observations[0].description == 'Unknown (Unknown Unknown)'
+
+    def test_result_round_trips_through_oscal_serialization(self, sample_blob):
+        """The Result must be genuinely OSCAL-schema-valid, not just pydantic-constructible."""
+        transformer = AwsConfigResultToOscalARTransformer()
+        results = transformer.transform(sample_blob)
+        serialized = results.oscal_serialize_json_bytes(pretty=True)
+        assert b'"aws-config-result"' not in serialized  # sanity: not leaking internal names
+        assert b'observations' in serialized
+        assert b'restricted-ssh' in serialized