From 4986d44e5f2fa6986b2546f1700865c0b36e92ac Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 2 Oct 2026 16:18:51 -0400 Subject: [PATCH] Enhance AES check with successful export history and passphrase warning --- aci-preupgrade-validation-script.py | 73 +++++++++-- docs/docs/validations.md | 9 ++ .../test_aes_encryption_check.py | 121 +++++++++++++++++- 3 files changed, 192 insertions(+), 11 deletions(-) diff --git a/aci-preupgrade-validation-script.py b/aci-preupgrade-validation-script.py index 5045dcd0..284e09b8 100644 --- a/aci-preupgrade-validation-script.py +++ b/aci-preupgrade-validation-script.py @@ -6436,21 +6436,41 @@ def equipment_disk_limits_exceeded(**kwargs): ) +def _config_export_age(timestamp): + """Calculate export age in UTC, including on APICs running Python 2.7.""" + match = re.match(r'^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.\d+)?(Z|[+-]\d{2}:\d{2})$', timestamp) + if not match: + raise ValueError("Invalid export completion timestamp") + completed = datetime.strptime(match.group(1), '%Y-%m-%dT%H:%M:%S') + offset = match.group(2) + if offset != 'Z': + hours, minutes = int(offset[1:3]), int(offset[4:6]) + if hours > 23 or minutes > 59: + raise ValueError("Invalid export completion timezone") + delta = timedelta(hours=hours, minutes=minutes) + completed = completed - delta if offset[0] == '+' else completed + delta + age = datetime.utcnow() - completed + if age.total_seconds() < 0: + return "In the future (check APIC clock)" + return '{}d {}h {}m'.format(age.days, age.seconds // 3600, age.seconds % 3600 // 60) + + @check_wrapper(check_title='Global AES Encryption') def aes_encryption_check(tversion, **kwargs): - result = FAIL_UF - headers = ["Target Version", "Global AES Encryption", "Impact"] - data = [] + headers = ["Target Version", "Global AES Encryption", "Impact", "Export Policy", "Last Successful Export", "Export Age"] recommended_action = ( - "\n\tEnable Global AES Encryption before upgrading your APIC (and take a configuration backup)." - "\n\tGlobal AES Encryption ensures that all configurations are included in the backup securely." + "\n\tEnsure Global AES Encryption is enabled before upgrading your APIC and a current configuration backup is available." + "\n\tGlobal AES Encryption allows secure properties to be included in configuration backups." + "\n\tWARNING: Ensure the AES encryption passphrase is known or saved in a known, secure location." + " It cannot be retrieved from APIC. AES-encrypted configuration exports cannot be restored" + " without the original passphrase. This check cannot verify that the customer knows or has saved it." ) doc_url = "https://datacenter.github.io/ACI-Pre-Upgrade-Validation-Script/validations/#global-aes-encryption" if not tversion: return Result(result=MANUAL, msg=TVER_MISSING) - if tversion.newer_than("6.1(2a)"): + if not tversion.older_than("6.1(2a)"): impact = "Upgrade Failure" result = FAIL_UF recommended_action += "\n\tUpgrade to 6.1(2) or later will fail when it is not enabled." @@ -6460,11 +6480,44 @@ def aes_encryption_check(tversion, **kwargs): cryptkeys = icurl("mo", "uni/exportcryptkey.json") if not cryptkeys: - data = [[str(tversion), "Object Not Found", impact]] - elif cryptkeys[0]["pkiExportEncryptionKey"]["attributes"]["strongEncryptionEnabled"] != "yes": - data = [[str(tversion), "Disabled", impact]] + result = MANUAL + encryption = "Object Not Found" + impact = "Unable to confirm encryption status" else: - result = PASS + enabled = cryptkeys[0]["pkiExportEncryptionKey"]["attributes"].get("strongEncryptionEnabled") + if enabled == "yes": + result = PASS + encryption, impact = "Enabled", "-" + elif enabled == "no": + encryption = "Disabled" + else: + result = ERROR + encryption = "Unknown" + impact = "Missing or unexpected strongEncryptionEnabled value" + + policy, completed, age = "-", "No successful export found in retained history", "-" + export_query = ( + 'configJob.json?query-target-filter=and(eq(configJob.type,"export"),eq(configJob.operSt,"success"))' + '&order-by=configJob.lastStepTime|desc' + ) + try: + # icurl() fetches every page. Only the first job is needed here. + exports = _icurl("class", export_query, page=0, page_size=1) + if int(exports["totalCount"]) > 0 and not exports["imdata"]: + raise ValueError("Export history response empty despite nonzero totalCount") + if exports["imdata"]: + attributes = exports["imdata"][0]["configJob"]["attributes"] + completed = attributes["lastStepTime"] + age = _config_export_age(completed) + policy_match = re.search(r'/jobs-\[uni/fabric/configexp-(.+)\]/run-', attributes["dn"]) + policy = policy_match.group(1) if policy_match else "Unknown" + except Exception: + log.warning("Unable to determine the latest successful configuration export", exc_info=True) + policy, completed, age = "-", "Unable to determine latest successful export", "-" + # Keep informational export details visible in APIC JSON even for PASS, + # whose failureDetails table is omitted by AciResult. + recommended_action += "\n\tExport history (informational): {} (policy: {}; age: {}).".format(completed, policy, age) + data = [[str(tversion), encryption, impact, policy, completed, age]] return Result(result=result, headers=headers, data=data, recommended_action=recommended_action, doc_url=doc_url) diff --git a/docs/docs/validations.md b/docs/docs/validations.md index 677a960a..fbf46d8b 100644 --- a/docs/docs/validations.md +++ b/docs/docs/validations.md @@ -2389,6 +2389,15 @@ When **Global AES Encryption** is not enabled, this script alerts users in two d * When it is not enabled and the target version is 6.1(2) or later, this check is flagged as `UPGRADE FAILURE`. * When it is not enabled and the target version is older than 6.1(2), this check is flagged as `MANUAL CHECK REQUIRED` to encourage users to follow the best practice to enable it (and take a configuration back again before the upgrade). +This check reads `strongEncryptionEnabled` from `pkiExportEncryptionKey` at `uni/exportcryptkey`. A missing object is flagged as `MANUAL CHECK REQUIRED` because encryption status cannot be confirmed. A missing or unexpected encryption attribute is flagged as `ERROR`. + +The check also reports the policy name, completion time, and age of the latest successful configuration export in retained APIC `configJob` history. It filters for `type="export"` and `operSt="success"`, sorts by `lastStepTime` descending, and reads only the first job. Export history is informational and does not change the AES result or impose a backup age threshold. `configExportP.triggerTime` is not used because it does not reliably reflect scheduled export runs. If there is no successful export in retained history, or history cannot be read, the check reports that explicitly. An export completion record does not establish that the backup file is still available or usable. + +!!! warning "Keep the AES encryption passphrase available" + Ensure the AES encryption passphrase is known or saved in a known, secure location. It cannot be retrieved from APIC. AES-encrypted configuration exports cannot be restored without the original passphrase. This check cannot verify that the customer knows or has saved the passphrase. + +This check reads configuration and export history only; it does not trigger, collect, or download a configuration export. + ### Service Graph BD Forceful Routing diff --git a/tests/checks/aes_encryption_check/test_aes_encryption_check.py b/tests/checks/aes_encryption_check/test_aes_encryption_check.py index 440a8603..55f9e67b 100644 --- a/tests/checks/aes_encryption_check/test_aes_encryption_check.py +++ b/tests/checks/aes_encryption_check/test_aes_encryption_check.py @@ -2,6 +2,7 @@ import pytest import logging import importlib +from datetime import datetime from helpers.utils import read_data script = importlib.import_module("aci-preupgrade-validation-script") @@ -14,6 +15,25 @@ # icurl queries exportcryptkey = "uni/exportcryptkey.json" +export_history = ( + 'configJob.json?query-target-filter=and(eq(configJob.type,"export"),eq(configJob.operSt,"success"))' + '&order-by=configJob.lastStepTime|desc' +) + + +@pytest.fixture(autouse=True) +def default_export_history(icurl_outputs): + icurl_outputs.setdefault(export_history, []) + + +@pytest.fixture +def fixed_time(monkeypatch): + class FixedDatetime(datetime): + @classmethod + def utcnow(cls): + return cls(2026, 10, 2, 14, 0, 0) + + monkeypatch.setattr(script, "datetime", FixedDatetime) @pytest.mark.parametrize( @@ -37,6 +57,17 @@ "6.1(3b)", script.FAIL_UF, ), + # Exact mandatory-encryption boundary + ( + {exportcryptkey: read_data(dir, "exportcryptkey_disabled.json")}, + "6.1(2a)", + script.FAIL_UF, + ), + ( + {exportcryptkey: read_data(dir, "exportcryptkey.json")}, + "6.1(2a)", + script.PASS, + ), # AES disabled (tversion < 6.1.2a) ( {exportcryptkey: read_data(dir, "exportcryptkey_disabled.json")}, @@ -47,7 +78,7 @@ ( {exportcryptkey: []}, "6.1(3b)", - script.FAIL_UF, + script.MANUAL, ), # AES MO not found (tversion < 6.1.2a) ( @@ -60,3 +91,91 @@ def test_logic(run_check, mock_icurl, tversion, expected_result): result = run_check(tversion=script.AciVersion(tversion)) assert result.result == expected_result + assert "No successful export found in retained history" in result.data[0] + assert "It cannot be retrieved from APIC" in result.recommended_action + assert "without the original passphrase" in result.recommended_action + + +@pytest.mark.parametrize("enabled, expected", [("yes", script.PASS), ("no", script.FAIL_UF), (None, script.ERROR), ("unknown", script.ERROR)]) +def test_encryption_attribute(run_check, mock_icurl, icurl_outputs, enabled, expected): + attributes = {} if enabled is None else {"strongEncryptionEnabled": enabled} + # No keyConfigured attribute is needed to evaluate encryption status. + icurl_outputs[exportcryptkey] = [{"pkiExportEncryptionKey": {"attributes": attributes}}] + result = run_check(tversion=script.AciVersion("6.1(3b)")) + assert result.result == expected + + +@pytest.mark.parametrize("enabled, expected", [("yes", script.PASS), ("no", script.FAIL_UF)]) +def test_latest_export_details(run_check, mock_icurl, icurl_outputs, monkeypatch, fixed_time, enabled, expected): + icurl_outputs[exportcryptkey] = [{"pkiExportEncryptionKey": {"attributes": {"strongEncryptionEnabled": enabled}}}] + icurl_outputs[export_history] = { + "totalCount": "20", + "imdata": [{"configJob": {"attributes": { + "dn": "uni/backupst/jobs-[uni/fabric/configexp-DailyAutoBackup]/run-2026-10-02T09-00-15", + "executeTime": "2026-10-02T09:00:15.685-04:00", + "lastStepTime": "2026-10-02T09:00:48.102-04:00", + "type": "export", + "operSt": "success", + }}}], + } + calls = [] + original_icurl = script._icurl + + def track_query(apitype, query, page=0, page_size=100000): + calls.append((apitype, query, page, page_size)) + return original_icurl(apitype, query, page=page, page_size=page_size) + + monkeypatch.setattr(script, "_icurl", track_query) + result = run_check(tversion=script.AciVersion("6.1(3b)")) + assert result.result == expected + assert result.data[0][3:] == ["DailyAutoBackup", "2026-10-02T09:00:48.102-04:00", "0d 0h 59m"] + assert len(result.headers) == len(result.data[0]) + assert calls == [("mo", exportcryptkey, 0, 100000), ("class", export_history, 0, 1)] + + output = [] + monkeypatch.setattr(script, "prints", output.append) + script.print_result(1, 1, "Global AES Encryption", **result.as_dict()) + assert "DailyAutoBackup" in output[0] + assert "0d 0h 59m" in output[0] + assert "WARNING:" in output[0] + payload = script.AciResult(test_function, "Global AES Encryption", result).as_dict() + assert "DailyAutoBackup" in payload["recommended_action"] + assert "2026-10-02T09:00:48.102-04:00" in payload["recommended_action"] + assert "without the original passphrase" in payload["recommended_action"] + + +@pytest.mark.parametrize("response", [ + {"totalCount": "1", "imdata": []}, + {"totalCount": "1", "imdata": [{"configJob": {"attributes": {"lastStepTime": "invalid"}}}]}, + {"totalCount": "1", "imdata": [{"configJob": {"attributes": {}}}]}, + {"totalCount": "1", "imdata": [{"error": {"attributes": {"text": "unresolved class for configJob"}}}]}, + {"totalCount": "1", "imdata": [{"error": {"attributes": {"text": "Unable to deliver the message, Resolve timeout"}}}]}, +]) +def test_export_history_errors_are_informational(run_check, mock_icurl, icurl_outputs, response): + icurl_outputs[exportcryptkey] = read_data(dir, "exportcryptkey.json") + icurl_outputs[export_history] = response + result = run_check(tversion=script.AciVersion("6.1(3b)")) + assert result.result == script.PASS + assert "Unable to determine latest successful export" in result.data[0] + assert "Unable to determine latest successful export" in result.recommended_action + assert "No successful export found" not in result.recommended_action + + +@pytest.mark.parametrize("timestamp, expected", [ + ("2026-10-01T09:00:00.000-04:00", "1d 1h 0m"), + ("2026-10-02T18:30:00.000+05:30", "0d 1h 0m"), + ("2026-10-02T13:00:00Z", "0d 1h 0m"), + ("2026-10-02T15:00:00.000+00:00", "In the future (check APIC clock)"), +]) +def test_export_age_handles_timezone_offsets(fixed_time, timestamp, expected): + assert script._config_export_age(timestamp) == expected + + +def test_missing_target_does_not_query_apic(run_check, monkeypatch): + def unexpected_query(*args, **kwargs): + pytest.fail("Missing target version must skip APIC queries") + + monkeypatch.setattr(script, "_icurl", unexpected_query) + result = run_check(tversion=None) + assert result.result == script.MANUAL + assert result.msg == script.TVER_MISSING