(record)
| 2032 | |
| 2033 | |
| 2034 | def _decrypt_creds(record): |
| 2035 | if not isinstance(record, dict): |
| 2036 | return None |
| 2037 | if record.get("__cred_store") == CRED_STORE_TAG_V2 and Fernet is not None: |
| 2038 | try: |
| 2039 | token = str(record.get("token") or "").encode("ascii") |
| 2040 | plain = Fernet(_fernet_key()).decrypt(token) |
| 2041 | out = json.loads(plain.decode("utf-8")) |
| 2042 | return out if isinstance(out, dict) else None |
| 2043 | except (InvalidToken, Exception): |
| 2044 | return None |
| 2045 | if record.get("__cred_store") != CRED_STORE_TAG: |
| 2046 | return None |
| 2047 | try: |
| 2048 | nonce_b64 = record.get("nonce", "") |
| 2049 | mac_b64 = record.get("mac", "") |
| 2050 | blob_b64 = record.get("blob", "") |
| 2051 | nonce = base64.urlsafe_b64decode(nonce_b64.encode("ascii", errors="ignore")) |
| 2052 | mac = base64.urlsafe_b64decode(mac_b64.encode("ascii", errors="ignore")) |
| 2053 | blob = base64.urlsafe_b64decode(blob_b64.encode("ascii", errors="ignore")) |
| 2054 | except Exception: |
| 2055 | return None |
| 2056 | if len(nonce) < 8 or len(blob) < 1: |
| 2057 | return None |
| 2058 | key = _credential_master_key() |
| 2059 | mask = _cred_xor_mask(key, nonce, len(blob)) |
| 2060 | plain = bytes(a ^ b for a, b in zip(blob, mask)) |
| 2061 | expected = hmac.new(key, nonce + plain, hashlib.sha256).digest() |
| 2062 | if not hmac.compare_digest(mac, expected): |
| 2063 | return None |
| 2064 | try: |
| 2065 | out = json.loads(plain.decode("utf-8")) |
| 2066 | except Exception: |
| 2067 | return None |
| 2068 | return out if isinstance(out, dict) else None |
| 2069 | |
| 2070 | |
| 2071 | def _load_credentials(): |
no test coverage detected