Authenticate using SCRAM.
(credentials: MongoCredential, conn: Connection, mechanism: str)
| 69 | |
| 70 | |
| 71 | def _authenticate_scram(credentials: MongoCredential, conn: Connection, mechanism: str) -> None: |
| 72 | """Authenticate using SCRAM.""" |
| 73 | username = credentials.username |
| 74 | if mechanism == "SCRAM-SHA-256": |
| 75 | digest = "sha256" |
| 76 | digestmod = hashlib.sha256 |
| 77 | data = saslprep(credentials.password).encode("utf-8") |
| 78 | else: |
| 79 | digest = "sha1" |
| 80 | digestmod = hashlib.sha1 |
| 81 | data = _password_digest(username, credentials.password).encode("utf-8") |
| 82 | source = credentials.source |
| 83 | cache = credentials.cache |
| 84 | |
| 85 | # Make local |
| 86 | _hmac = hmac.HMAC |
| 87 | |
| 88 | ctx = conn.auth_ctx |
| 89 | if ctx and ctx.speculate_succeeded(): |
| 90 | assert isinstance(ctx, _ScramContext) |
| 91 | assert ctx.scram_data is not None |
| 92 | nonce, first_bare = ctx.scram_data |
| 93 | res = ctx.speculative_authenticate |
| 94 | else: |
| 95 | nonce, first_bare, cmd = _authenticate_scram_start(credentials, mechanism) |
| 96 | res = conn.command(source, cmd) |
| 97 | |
| 98 | assert res is not None |
| 99 | server_first = res["payload"] |
| 100 | parsed = _parse_scram_response(server_first) |
| 101 | iterations = int(parsed[b"i"]) |
| 102 | if iterations < 4096: |
| 103 | raise OperationFailure("Server returned an invalid iteration count.") |
| 104 | salt = parsed[b"s"] |
| 105 | rnonce = parsed[b"r"] |
| 106 | if not rnonce.startswith(nonce): |
| 107 | raise OperationFailure("Server returned an invalid nonce.") |
| 108 | |
| 109 | without_proof = b"c=biws,r=" + rnonce |
| 110 | if cache.data: |
| 111 | client_key, server_key, csalt, citerations = cache.data |
| 112 | else: |
| 113 | client_key, server_key, csalt, citerations = None, None, None, None |
| 114 | |
| 115 | # Salt and / or iterations could change for a number of different |
| 116 | # reasons. Either changing invalidates the cache. |
| 117 | if not client_key or salt != csalt or iterations != citerations: |
| 118 | salted_pass = hashlib.pbkdf2_hmac(digest, data, standard_b64decode(salt), iterations) |
| 119 | client_key = _hmac(salted_pass, b"Client Key", digestmod).digest() |
| 120 | server_key = _hmac(salted_pass, b"Server Key", digestmod).digest() |
| 121 | cache.data = (client_key, server_key, salt, iterations) |
| 122 | stored_key = digestmod(client_key).digest() |
| 123 | auth_msg = b",".join((first_bare, server_first, without_proof)) |
| 124 | client_sig = _hmac(stored_key, auth_msg, digestmod).digest() |
| 125 | client_proof = b"p=" + standard_b64encode(_xor(client_key, client_sig)) |
| 126 | client_final = b",".join((without_proof, client_proof)) |
| 127 | |
| 128 | server_sig = standard_b64encode(_hmac(server_key, auth_msg, digestmod).digest()) |
no test coverage detected