(cls, kvp)
| 141 | |
| 142 | @classmethod |
| 143 | def to_model(cls, kvp): |
| 144 | if not KeyValuePairAPI.crypto_setup: |
| 145 | KeyValuePairAPI._setup_crypto() |
| 146 | |
| 147 | kvp_id = getattr(kvp, "id", None) |
| 148 | name = getattr(kvp, "name", None) |
| 149 | description = getattr(kvp, "description", None) |
| 150 | value = kvp.value |
| 151 | original_value = value |
| 152 | secret = False |
| 153 | |
| 154 | if getattr(kvp, "ttl", None): |
| 155 | expire_timestamp = date_utils.get_datetime_utc_now() + datetime.timedelta( |
| 156 | seconds=kvp.ttl |
| 157 | ) |
| 158 | else: |
| 159 | expire_timestamp = None |
| 160 | |
| 161 | encrypted = getattr(kvp, "encrypted", False) |
| 162 | secret = getattr(kvp, "secret", False) |
| 163 | |
| 164 | # If user transmitted the value in an pre-encrypted format, we perform the decryption here |
| 165 | # to ensure data integrity. Besides that, we store data as-is. |
| 166 | # Keep in mind that encrypted=True also always implies secret=True. If we didn't do |
| 167 | # that and supported encrypted=True, secret=False, this would allow users to decrypt |
| 168 | # any encrypted value. |
| 169 | if encrypted: |
| 170 | secret = True |
| 171 | |
| 172 | cls._verif_key_is_set_up(name=name) |
| 173 | |
| 174 | try: |
| 175 | symmetric_decrypt(KeyValuePairAPI.crypto_key, value) |
| 176 | except Exception: |
| 177 | msg = ( |
| 178 | 'Failed to verify the integrity of the provided value for key "%s". Ensure ' |
| 179 | "that the value is encrypted with the correct key and not corrupted." |
| 180 | % (name) |
| 181 | ) |
| 182 | raise ValueError(msg) |
| 183 | |
| 184 | # Additional safety check to ensure that the value hasn't been decrypted |
| 185 | if value != original_value: |
| 186 | raise ValueError( |
| 187 | f"The encrypted value {value} is not the" |
| 188 | f" same original encrypted value {original_value}." |
| 189 | ) |
| 190 | elif secret: |
| 191 | cls._verif_key_is_set_up(name=name) |
| 192 | |
| 193 | value = symmetric_encrypt(KeyValuePairAPI.crypto_key, value) |
| 194 | |
| 195 | scope = getattr(kvp, "scope", FULL_SYSTEM_SCOPE) |
| 196 | |
| 197 | if scope not in ALLOWED_SCOPES: |
| 198 | raise InvalidScopeException( |
| 199 | 'Invalid scope "%s"! Allowed scopes are %s.' % (scope, ALLOWED_SCOPES) |
| 200 | ) |
nothing calls this directly
no test coverage detected