Process all instance metadata cleaning it up for persisting as json. Strip ci-b64 prefix and catalog any 'base64_encoded_keys' as a list @return Dict copy of processed metadata.
(metadata, key_path="", sensitive_keys=())
| 113 | |
| 114 | |
| 115 | def process_instance_metadata(metadata, key_path="", sensitive_keys=()): |
| 116 | """Process all instance metadata cleaning it up for persisting as json. |
| 117 | |
| 118 | Strip ci-b64 prefix and catalog any 'base64_encoded_keys' as a list |
| 119 | |
| 120 | @return Dict copy of processed metadata. |
| 121 | """ |
| 122 | md_copy = copy.deepcopy(metadata) |
| 123 | base64_encoded_keys = [] |
| 124 | sens_keys = [] |
| 125 | for key, val in metadata.items(): |
| 126 | if key_path: |
| 127 | sub_key_path = key_path + "/" + key |
| 128 | else: |
| 129 | sub_key_path = key |
| 130 | if ( |
| 131 | key.lower() in sensitive_keys |
| 132 | or sub_key_path.lower() in sensitive_keys |
| 133 | ): |
| 134 | sens_keys.append(sub_key_path) |
| 135 | if isinstance(val, str) and val.startswith("ci-b64:"): |
| 136 | base64_encoded_keys.append(sub_key_path) |
| 137 | md_copy[key] = val.replace("ci-b64:", "") |
| 138 | if isinstance(val, dict): |
| 139 | return_val = process_instance_metadata( |
| 140 | val, sub_key_path, sensitive_keys |
| 141 | ) |
| 142 | base64_encoded_keys.extend(return_val.pop("base64_encoded_keys")) |
| 143 | sens_keys.extend(return_val.pop("sensitive_keys")) |
| 144 | md_copy[key] = return_val |
| 145 | md_copy["base64_encoded_keys"] = sorted(base64_encoded_keys) |
| 146 | md_copy["sensitive_keys"] = sorted(sens_keys) |
| 147 | return md_copy |
| 148 | |
| 149 | |
| 150 | def redact_sensitive_keys(metadata, redact_value=REDACT_SENSITIVE_VALUE): |
no test coverage detected