expose credentials expiration (#219)

Author: graebm

awscrt/auth.py

@@ -24,22 +24,47 @@ class AwsCredentials(NativeResource):
     Args:
         access_key_id (str): Access key ID
         secret_access_key (str): Secret access key
-        session_token (Optional[str]): Session token
+        session_token (Optional[str]): Optional security token associated with
+            the credentials.
+        expiration (Optional[datetime.datetime]): Optional expiration datetime,
+            that the credentials will no longer be valid past.
+            Converted to UTC timezone and rounded down to nearest second.
+            If not set, then credentials do not expire.
 
     Attributes:
         access_key_id (str): Access key ID
         secret_access_key (str): Secret access key
-        session_token (Optional[str]): Session token
+        session_token (Optional[str]): Security token associated with
+            the credentials. None if not set.
+        expiration (Optional[datetime.datetime]): Expiration datetime,
+            that the credentials will no longer be valid past.
+            None if credentials do not expire.
+            Timezone is always UTC.
     """
     __slots__ = ()
 
-    def __init__(self, access_key_id, secret_access_key, session_token=None):
+    # C layer uses UINT64_MAX as timestamp for non-expiring credentials
+    _NONEXPIRING_TIMESTAMP = 0xFFFFFFFFFFFFFFFF
+
+    def __init__(self, access_key_id, secret_access_key, session_token=None, expiration=None):
         assert isinstance(access_key_id, str)
         assert isinstance(secret_access_key, str)
         assert isinstance(session_token, str) or session_token is None
 
+        # C layer uses large int as timestamp for non-expiring credentials
+        if expiration is None:
+            expiration_timestamp = self._NONEXPIRING_TIMESTAMP
+        else:
+            expiration_timestamp = int(expiration.timestamp())
+            if expiration_timestamp < 0 or expiration_timestamp >= self._NONEXPIRING_TIMESTAMP:
+                raise OverflowError("expiration datetime out of range")
+
         super().__init__()
-        self._binding = _awscrt.credentials_new(access_key_id, secret_access_key, session_token)
+        self._binding = _awscrt.credentials_new(
+            access_key_id,
+            secret_access_key,
+            session_token,
+            expiration_timestamp)
 
     @classmethod
     def _from_binding(cls, binding):
@@ -61,6 +86,15 @@ def secret_access_key(self):
     def session_token(self):
         return _awscrt.credentials_session_token(self._binding)
 
+    @property
+    def expiration(self):
+        timestamp = _awscrt.credentials_expiration_timestamp_seconds(self._binding)
+        # C layer uses large int as timestamp for non-expiring credentials
+        if timestamp == self._NONEXPIRING_TIMESTAMP:
+            return None
+        else:
+            return datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc)
+
     def __deepcopy__(self, memo):
         # AwsCredentials is immutable, so just return self.
         return self

source/auth.h

@@ -12,6 +12,7 @@ PyObject *aws_py_credentials_new(PyObject *self, PyObject *args);
 PyObject *aws_py_credentials_access_key_id(PyObject *self, PyObject *args);
 PyObject *aws_py_credentials_secret_access_key(PyObject *self, PyObject *args);
 PyObject *aws_py_credentials_session_token(PyObject *self, PyObject *args);
+PyObject *aws_py_credentials_expiration_timestamp_seconds(PyObject *self, PyObject *args);
 
 PyObject *aws_py_credentials_provider_get_credentials(PyObject *self, PyObject *args);
 PyObject *aws_py_credentials_provider_new_chain_default(PyObject *self, PyObject *args);

source/auth_credentials.c

@@ -26,24 +26,22 @@ PyObject *aws_py_credentials_new(PyObject *self, PyObject *args) {
     struct aws_byte_cursor access_key_id;
     struct aws_byte_cursor secret_access_key;
     struct aws_byte_cursor session_token; /* session_token is optional */
+    uint64_t expiration_timestamp_sec;
     if (!PyArg_ParseTuple(
             args,
-            "s#s#z#",
+            "s#s#z#K",
             &access_key_id.ptr,
             &access_key_id.len,
             &secret_access_key.ptr,
             &secret_access_key.len,
             &session_token.ptr,
-            &session_token.len)) {
+            &session_token.len,
+            &expiration_timestamp_sec)) {
         return NULL;
     }
 
     struct aws_credentials *credentials = aws_credentials_new(
-        aws_py_get_allocator(),
-        access_key_id,
-        secret_access_key,
-        session_token,
-        UINT64_MAX /*expiration_timepoint_seconds*/);
+        aws_py_get_allocator(), access_key_id, secret_access_key, session_token, expiration_timestamp_sec);
     if (!credentials) {
         return PyErr_AwsLastError();
     }
@@ -115,6 +113,23 @@ PyObject *aws_py_credentials_session_token(PyObject *self, PyObject *args) {
     return s_credentials_get_member_str(args, CREDENTIALS_MEMBER_SESSION_TOKEN);
 }
 
+PyObject *aws_py_credentials_expiration_timestamp_seconds(PyObject *self, PyObject *args) {
+    (void)self;
+
+    PyObject *capsule;
+    if (!PyArg_ParseTuple(args, "O", &capsule)) {
+        return NULL;
+    }
+
+    const struct aws_credentials *credentials = PyCapsule_GetPointer(capsule, s_capsule_name_credentials);
+    if (!credentials) {
+        return NULL;
+    }
+
+    uint64_t timestamp = aws_credentials_get_expiration_timepoint_seconds(credentials);
+    return PyLong_FromUnsignedLongLong(timestamp);
+}
+
 /**
  * Binds a Python CredentialsProvider to a native aws_credentials_provider.
  */

source/module.c

@@ -543,6 +543,7 @@ static PyMethodDef s_module_methods[] = {
     AWS_PY_METHOD_DEF(credentials_access_key_id, METH_VARARGS),
     AWS_PY_METHOD_DEF(credentials_secret_access_key, METH_VARARGS),
     AWS_PY_METHOD_DEF(credentials_session_token, METH_VARARGS),
+    AWS_PY_METHOD_DEF(credentials_expiration_timestamp_seconds, METH_VARARGS),
     AWS_PY_METHOD_DEF(credentials_provider_get_credentials, METH_VARARGS),
     AWS_PY_METHOD_DEF(credentials_provider_new_chain_default, METH_VARARGS),
     AWS_PY_METHOD_DEF(credentials_provider_new_static, METH_VARARGS),

test/test_auth.py

@@ -13,6 +13,7 @@
 EXAMPLE_ACCESS_KEY_ID = 'example_access_key_id'
 EXAMPLE_SECRET_ACCESS_KEY = 'example_secret_access_key'
 EXAMPLE_SESSION_TOKEN = 'example_session_token'
+EXAMPLE_SESSION_EXPIRATION = datetime.datetime.fromtimestamp(1609911816, tz=datetime.timezone.utc)
 
 
 class ScopedEnvironmentVariable:
@@ -40,12 +41,14 @@ def test_create(self):
         credentials = awscrt.auth.AwsCredentials(
             EXAMPLE_ACCESS_KEY_ID,
             EXAMPLE_SECRET_ACCESS_KEY,
-            EXAMPLE_SESSION_TOKEN)
+            EXAMPLE_SESSION_TOKEN,
+            EXAMPLE_SESSION_EXPIRATION)
 
         # Don't use assertEqual(), which could log actual credentials if test fails.
         self.assertTrue(EXAMPLE_ACCESS_KEY_ID == credentials.access_key_id)
         self.assertTrue(EXAMPLE_SECRET_ACCESS_KEY == credentials.secret_access_key)
         self.assertTrue(EXAMPLE_SESSION_TOKEN == credentials.session_token)
+        self.assertTrue(EXAMPLE_SESSION_EXPIRATION == credentials.expiration)
 
     def test_create_no_session_token(self):
         credentials = awscrt.auth.AwsCredentials(EXAMPLE_ACCESS_KEY_ID, EXAMPLE_SECRET_ACCESS_KEY)
@@ -55,6 +58,14 @@ def test_create_no_session_token(self):
         self.assertTrue(EXAMPLE_SECRET_ACCESS_KEY == credentials.secret_access_key)
         self.assertTrue(credentials.session_token is None)
 
+    def test_create_no_expiration(self):
+        credentials = awscrt.auth.AwsCredentials(EXAMPLE_ACCESS_KEY_ID, EXAMPLE_SECRET_ACCESS_KEY)
+
+        # Don't use assertEqual(), which could log actual credentials if test fails.
+        self.assertTrue(EXAMPLE_ACCESS_KEY_ID == credentials.access_key_id)
+        self.assertTrue(EXAMPLE_SECRET_ACCESS_KEY == credentials.secret_access_key)
+        self.assertTrue(credentials.expiration is None)
+
 
 class TestProvider(NativeResourceTest):
     def test_static_provider(self):