duration_seconds

Author: Rishang

broker/aws/sts.py

@@ -32,12 +32,20 @@
 
 
 class AWSRoleSTS:
-    def __init__(self, role_arn: str, username: str = "", region: str = None) -> None:
+    def __init__(
+        self,
+        role_arn: str,
+        username: str = "",
+        region: str = None,
+        duration_seconds: int = 3600,
+    ) -> None:
+
         self.role_arn = role_arn
         self.username = username
         self.region = region
+        self.duration_seconds = duration_seconds
 
-    def oidc_sts(self, jwt_token: str, duration_seconds: int = 3600) -> dict:
+    def oidc_sts(self, jwt_token: str) -> dict:
         """
         Returns a boto3 client for OpenID Connect STS (Security Token Service)
         """
@@ -49,11 +57,9 @@ def oidc_sts(self, jwt_token: str, duration_seconds: int = 3600) -> dict:
             RoleArn=self.role_arn,
             RoleSessionName=self.username or "sts-role-session",
             WebIdentityToken=jwt_token,
-            DurationSeconds=duration_seconds,
+            DurationSeconds=self.duration_seconds,
         )
 
-        self.duration_seconds = duration_seconds
-
         if isinstance(self.region, str) or self.region in aws_regions:
             # login based on region if provided
             self.response["Region"] = self.region
@@ -124,13 +130,23 @@ def quote_plus_function(s):
 
 
 def get_role(
-    token, role: str, username: str = "", issuer: str = None, region: str = None
+    token,
+    role: str,
+    username: str = "",
+    issuer: str = None,
+    region: str = None,
+    duration_seconds: int = 3600,
 ):
     """Provide aws sts role access to aws cli or console based on web identity token"""
 
     sts: dict = {}
     if isinstance(username, str) and username != "":
-        aws_role = AWSRoleSTS(role_arn=role, username=username, region=region)
+        aws_role = AWSRoleSTS(
+            role_arn=role,
+            username=username,
+            region=region,
+            duration_seconds=duration_seconds,
+        )
     else:
         aws_role = AWSRoleSTS(role_arn=role)
     try:

broker/routes.py

@@ -74,11 +74,12 @@ def aws_auth():
     """provide aws iam role based creds to"""
 
     args: dict = request.args.to_dict()
+
     role: str = args.get("role")
     region: str = args.get("region")
-
     userinfo = session.get("user")
     type: str = args.get("type")
+    duration_s: str = args.get("duration_seconds") or 3600
 
     if None in [userinfo, role]:
         return redirect(url_for("login", role=role))
@@ -95,7 +96,9 @@ def aws_auth():
         username=userinfo.get("preferred_username"),
         issuer=request.headers.get("Host"),
         region=region,
+        duration_seconds=int(duration_s),
     )
+
     if sts_role["expired"] == True:
         return redirect(url_for("login"))