src/mount_efs/__init__.py [3078:3193]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - key_line = key_line.replace(" ", "") # DER encoding TLV (Tag, Length, Value) # - the first octet (byte) is the tag (type) # - the next octets are the length - "definite form" # - the first octet always has the high order bit (8) set to 1 # - the remaining 127 bits are used to encode the number of octets that follow # - the following octets encode, as big-endian, the length (which may be 0) as a number of octets # - the remaining octets are the "value" aka content # # For a BIT STRING, the first octet of the value is used to signify the number of unused bits that exist in the last # content byte. Note that this is explicitly excluded from the SubjectKeyIdentifier hash, per # https://tools.ietf.org/html/rfc5280#section-4.2.1.2 # # Example: # 0382018f00... # - 03 - BIT STRING tag # - 82 - 2 length octets to follow (ignore high order bit) # - 018f - length of 399 # - 00 - no unused bits in the last content byte offset = int(key_line.split(":")[0]) key = key[offset:] num_length_octets = key[1] & 0b01111111 # Exclude the tag (1), length (1 + num_length_octets), and number of unused bits (1) offset = 1 + 1 + num_length_octets + 1 key = key[offset:] sha1 = hashlib.sha1() sha1.update(key) return sha1.hexdigest() def create_canonical_request( public_key_hash, date, access_key, region, fs_id, session_token=None ): """ Create a Canonical Request - https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html """ formatted_datetime = date.strftime(SIGV4_DATETIME_FORMAT) credential = quote_plus(access_key + "/" + get_credential_scope(date, region)) request = HTTP_REQUEST_METHOD + "\n" request += CANONICAL_URI + "\n" request += ( create_canonical_query_string( public_key_hash, credential, formatted_datetime, session_token ) + "\n" ) request += CANONICAL_HEADERS % fs_id + "\n" request += SIGNED_HEADERS + "\n" sha256 = hashlib.sha256() sha256.update(REQUEST_PAYLOAD.encode()) request += sha256.hexdigest() return request def create_canonical_query_string( public_key_hash, credential, formatted_datetime, session_token=None ): canonical_query_params = { "Action": "Connect", # Public key hash is included in canonical request to tie the signature to a specific key pair to avoid replay attacks "PublicKeyHash": quote_plus(public_key_hash), "X-Amz-Algorithm": ALGORITHM, "X-Amz-Credential": credential, "X-Amz-Date": quote_plus(formatted_datetime), "X-Amz-Expires": 86400, "X-Amz-SignedHeaders": SIGNED_HEADERS, } if session_token: canonical_query_params["X-Amz-Security-Token"] = quote_plus(session_token) # Cannot use urllib.urlencode because it replaces the %s's return "&".join( ["%s=%s" % (k, v) for k, v in sorted(canonical_query_params.items())] ) def create_string_to_sign(canonical_request, date, region): """ Create a String to Sign - https://docs.aws.amazon.com/general/latest/gr/sigv4-create-string-to-sign.html """ string_to_sign = ALGORITHM + "\n" string_to_sign += date.strftime(SIGV4_DATETIME_FORMAT) + "\n" string_to_sign += get_credential_scope(date, region) + "\n" sha256 = hashlib.sha256() sha256.update(canonical_request.encode()) string_to_sign += sha256.hexdigest() return string_to_sign def calculate_signature(string_to_sign, date, secret_access_key, region): """ Calculate the Signature - https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html """ def _sign(key, msg): return hmac.new(key, msg.encode("utf-8"), hashlib.sha256) key_date = _sign( ("AWS4" + secret_access_key).encode("utf-8"), date.strftime(DATE_ONLY_FORMAT) ).digest() add_region = _sign(key_date, region).digest() add_service = _sign(add_region, SERVICE).digest() signing_key = _sign(add_service, "aws4_request").digest() return _sign(signing_key, string_to_sign).hexdigest() - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - src/watchdog/__init__.py [1977:2092]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - key_line = key_line.replace(" ", "") # DER encoding TLV (Tag, Length, Value) # - the first octet (byte) is the tag (type) # - the next octets are the length - "definite form" # - the first octet always has the high order bit (8) set to 1 # - the remaining 127 bits are used to encode the number of octets that follow # - the following octets encode, as big-endian, the length (which may be 0) as a number of octets # - the remaining octets are the "value" aka content # # For a BIT STRING, the first octet of the value is used to signify the number of unused bits that exist in the last # content byte. Note that this is explicitly excluded from the SubjectKeyIdentifier hash, per # https://tools.ietf.org/html/rfc5280#section-4.2.1.2 # # Example: # 0382018f00... # - 03 - BIT STRING tag # - 82 - 2 length octets to follow (ignore high order bit) # - 018f - length of 399 # - 00 - no unused bits in the last content byte offset = int(key_line.split(":")[0]) key = key[offset:] num_length_octets = key[1] & 0b01111111 # Exclude the tag (1), length (1 + num_length_octets), and number of unused bits (1) offset = 1 + 1 + num_length_octets + 1 key = key[offset:] sha1 = hashlib.sha1() sha1.update(key) return sha1.hexdigest() def create_canonical_request( public_key_hash, date, access_key, region, fs_id, session_token=None ): """ Create a Canonical Request - https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html """ formatted_datetime = date.strftime(SIGV4_DATETIME_FORMAT) credential = quote_plus(access_key + "/" + get_credential_scope(date, region)) request = HTTP_REQUEST_METHOD + "\n" request += CANONICAL_URI + "\n" request += ( create_canonical_query_string( public_key_hash, credential, formatted_datetime, session_token ) + "\n" ) request += CANONICAL_HEADERS % fs_id + "\n" request += SIGNED_HEADERS + "\n" sha256 = hashlib.sha256() sha256.update(REQUEST_PAYLOAD.encode()) request += sha256.hexdigest() return request def create_canonical_query_string( public_key_hash, credential, formatted_datetime, session_token=None ): canonical_query_params = { "Action": "Connect", # Public key hash is included in canonical request to tie the signature to a specific key pair to avoid replay attacks "PublicKeyHash": quote_plus(public_key_hash), "X-Amz-Algorithm": ALGORITHM, "X-Amz-Credential": credential, "X-Amz-Date": quote_plus(formatted_datetime), "X-Amz-Expires": 86400, "X-Amz-SignedHeaders": SIGNED_HEADERS, } if session_token: canonical_query_params["X-Amz-Security-Token"] = quote_plus(session_token) # Cannot use urllib.urlencode because it replaces the %s's return "&".join( ["%s=%s" % (k, v) for k, v in sorted(canonical_query_params.items())] ) def create_string_to_sign(canonical_request, date, region): """ Create a String to Sign - https://docs.aws.amazon.com/general/latest/gr/sigv4-create-string-to-sign.html """ string_to_sign = ALGORITHM + "\n" string_to_sign += date.strftime(SIGV4_DATETIME_FORMAT) + "\n" string_to_sign += get_credential_scope(date, region) + "\n" sha256 = hashlib.sha256() sha256.update(canonical_request.encode()) string_to_sign += sha256.hexdigest() return string_to_sign def calculate_signature(string_to_sign, date, secret_access_key, region): """ Calculate the Signature - https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html """ def _sign(key, msg): return hmac.new(key, msg.encode("utf-8"), hashlib.sha256) key_date = _sign( ("AWS4" + secret_access_key).encode("utf-8"), date.strftime(DATE_ONLY_FORMAT) ).digest() add_region = _sign(key_date, region).digest() add_service = _sign(add_region, SERVICE).digest() signing_key = _sign(add_service, "aws4_request").digest() return _sign(signing_key, string_to_sign).hexdigest() - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -