AT-14951: migrate from pylint/isort/black to ruff

.pre-commit-config.yaml

@@ -37,16 +37,12 @@ repos:
     hooks:
       - id: mdformat
         name: Format markdown
-  - repo: https://github.com/pycqa/isort
-    rev: 8.0.1
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.15.21
     hooks:
-      - id: isort
-        name: isort (python)
-        args: [--profile, black, --filter-files]
-  - repo: https://github.com/psf/black
-    rev: 26.3.1
-    hooks:
-      - id: black
+      - id: ruff-check
+        args: [--fix]
+      - id: ruff-format
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.19.1
     hooks:
@@ -55,10 +51,3 @@ repos:
         additional_dependencies:
           - types-requests
           - boto3-stubs[sts,iam]
-  - repo: https://github.com/pylint-dev/pylint
-    rev: v4.0.5
-    hooks:
-      - id: pylint
-        name: pylint (Python Linting)
-        files: ^(oktaawscli|tests)/
-        args: [--rcfile=pylintrc, --output-format=colorized, --score=no]

CONTRIBUTING.md

@@ -20,10 +20,7 @@ It's highly recommended to use virtualenv!
 pip install .
 ```
 
-- Ensure that you can run pylint against your code and no errors are returned. Pull Requests with pylint errors will be rejected.
-
-  - Currently, automated builds are only checking for actual errors, as there are some refactoring and other such notices that need to be resolved.
-  - You can safely run `pylint --errors-only oktaawscli` to replicate what the build will be checking.
+- Ensure `pre-commit run --all-files` passes (runs ruff for linting/formatting and mypy for type checking). Pull Requests with lint or type errors will be rejected.
 
 - Increment the version in `oktaawscli/version.py`, according to [SemVer](https://semver.org/).
 

oktaawscli/__init__.py

@@ -1,3 +1,3 @@
 """init"""
 
-from .version import __version__
+from .version import __version__ as __version__

oktaawscli/aws_auth.py

@@ -68,10 +68,7 @@ def choose_aws_role(self, assertion):
                 self.logger.info("Using predefined role: %s" % self.role)
                 return predefined_role
             else:
-                self.logger.info(
-                    "Predefined role, %s, not found in the list of roles assigned to you."
-                    % self.role
-                )
+                self.logger.info("Predefined role, %s, not found in the list of roles assigned to you." % self.role)
                 self.logger.info("Please choose a role.")
 
         if len(roles) == 1:
@@ -106,9 +103,7 @@ def get_sts_token(self, role_arn, principal_arn, assertion, duration):
             if profile is not None:
                 os.environ["AWS_PROFILE"] = profile
         except ProfileNotFound:
-            self.logger.exception(
-                "Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"]
-            )
+            self.logger.exception("Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"])
 
         response = sts.assume_role_with_saml(
             RoleArn=role_arn,
@@ -137,9 +132,7 @@ def check_sts_token(self, profile):
             return False
 
         elif not parser.has_section(profile):
-            self.logger.info(
-                "No existing credentials found. Requesting new credentials."
-            )
+            self.logger.info("No existing credentials found. Requesting new credentials.")
             return False
 
         session = boto3.Session(profile_name=profile)
@@ -149,23 +142,17 @@ def check_sts_token(self, profile):
 
         except (ClientError, NoCredentialsError) as ex:
             if str(ex) == "Unable to locate credentials":
-                self.logger.info(
-                    "No credentials have been located. Requesting new credentials."
-                )
+                self.logger.info("No credentials have been located. Requesting new credentials.")
                 return False
             elif ex.response["Error"]["Code"] == "ExpiredToken":
-                self.logger.info(
-                    "Temporary credentials have expired. Requesting new credentials."
-                )
+                self.logger.info("Temporary credentials have expired. Requesting new credentials.")
                 return False
 
         print("AWS credentials are valid. Nothing to do.")
         self.logger.info("STS credentials are valid. Nothing to do.")
         return True
 
-    def write_sts_token(
-        self, profile, access_key_id, secret_access_key, session_token, region=None
-    ):
+    def write_sts_token(self, profile, access_key_id, secret_access_key, session_token, region=None):
         """Writes STS auth information to credentials file"""
         region = region or self.region
         output = "json"
@@ -225,9 +212,7 @@ def __extract_available_roles_from(assertion):
         roles = []
         role_tuple = namedtuple("RoleTuple", ["principal_arn", "role_arn"])
         root = ET.fromstring(base64.b64decode(assertion))
-        for saml2attribute in root.iter(
-            "{urn:oasis:names:tc:SAML:2.0:assertion}Attribute"
-        ):
+        for saml2attribute in root.iter("{urn:oasis:names:tc:SAML:2.0:assertion}Attribute"):
             if saml2attribute.get("Name") == aws_attribute_role:
                 for saml2attributevalue in saml2attribute.iter(attribute_value_urn):
                     roles.append(role_tuple(*saml2attributevalue.text.split(",")))
@@ -257,12 +242,8 @@ def __get_role_info(self, roles, assertion):
                 current_date = date.today()
                 alias_age = current_date - last_updated
                 if alias_age.days >= 7 or alias is None:
-                    self.logger.info(
-                        "Refreshing cached alias for role %s" % role.role_arn
-                    )
-                    alias = self.__get_account_alias(
-                        role.role_arn, role.principal_arn, assertion
-                    )
+                    self.logger.info("Refreshing cached alias for role %s" % role.role_arn)
+                    alias = self.__get_account_alias(role.role_arn, role.principal_arn, assertion)
                     last_updated = current_date
                     if alias is None:
                         continue
@@ -301,13 +282,9 @@ def __get_account_alias(self, role_arn, principal_arn, assertion):
         try:
             sts = boto3.client("sts")
         except ProfileNotFound:
-            self.logger.exception(
-                "Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"]
-            )
+            self.logger.exception("Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"])
         try:
-            saml_resp = sts.assume_role_with_saml(
-                RoleArn=role_arn, PrincipalArn=principal_arn, SAMLAssertion=assertion
-            )
+            saml_resp = sts.assume_role_with_saml(RoleArn=role_arn, PrincipalArn=principal_arn, SAMLAssertion=assertion)
         except ClientError:
             self.logger.warning(
                 "Unable to assume role '%s', cannot get account alias",
@@ -337,19 +314,15 @@ def __get_account_alias(self, role_arn, principal_arn, assertion):
                     exc_info=self.debug,
                 )
             else:
-                self.logger.exception(
-                    "Unknown Error. Unable to get account alias for role %s", role_arn
-                )
+                self.logger.exception("Unknown Error. Unable to get account alias for role %s", role_arn)
             return "unknown"
 
     @staticmethod
     def __create_options_from(roles):
         options = []
         for index, role in enumerate(roles):
             # role[0] is the role arn, role[2] is the account alias
-            options.append(
-                "[%s]: %s : %s" % (str(index + 1).ljust(2), role[2].ljust(27), role[0])
-            )
+            options.append("[%s]: %s : %s" % (str(index + 1).ljust(2), role[2].ljust(27), role[0]))
         return options
 
     def __find_predefined_role_from(self, roles):

oktaawscli/okta_auth.py

@@ -25,9 +25,7 @@
 class OktaAuth:
     """Handles auth to Okta and returns SAML assertion"""
 
-    def __init__(
-        self, okta_profile, verbose, logger, totp_token, okta_auth_config, debug=False
-    ):
+    def __init__(self, okta_profile, verbose, logger, totp_token, okta_auth_config, debug=False):
         self.okta_profile = okta_profile
         self.totp_token = totp_token
         self.logger = logger
@@ -53,37 +51,28 @@ def primary_auth(self):
         with locked(self.token_path, timeout=INTERACTIVE_LOCK_TIMEOUT_SECONDS):
             refreshed = self.get_cached_session_id()
             if refreshed is not None and refreshed != session_id:
-                self.logger.info(
-                    "Cached Okta session was refreshed by another process; using it."
-                )
+                self.logger.info("Cached Okta session was refreshed by another process; using it.")
                 return refreshed
             return self.get_session(self._run_authn_flow())
 
     def _run_authn_flow(self):
         """Runs the Okta authn POST and returns a sessionToken. Caller holds the lock."""
-        self.logger.warning(
-            "Cached Okta session is missing or invalid. Authenticating now..."
-        )
+        self.logger.warning("Cached Okta session is missing or invalid. Authenticating now...")
         auth_data = {
             "username": self.okta_auth_config.username_for(self.okta_profile),
             "password": self.okta_auth_config.password_for(self.okta_profile),
         }
         # https://developer.okta.com/docs/reference/api/authn/
-        resp_json = self._okta_json_request(
-            "POST", "/api/v1/authn", "_run_authn_flow", json=auth_data
-        )
+        resp_json = self._okta_json_request("POST", "/api/v1/authn", "_run_authn_flow", json=auth_data)
         if "status" in resp_json:
             status = resp_json["status"]
             if status == "MFA_REQUIRED":
-                return self.verify_mfa(
-                    resp_json["_embedded"]["factors"], resp_json["stateToken"]
-                )
+                return self.verify_mfa(resp_json["_embedded"]["factors"], resp_json["stateToken"])
             if status == "SUCCESS":
                 return resp_json["sessionToken"]
             if status == "MFA_ENROLL":
                 self.logger.warning(
-                    "MFA not enrolled. Cannot continue. "
-                    "Please enroll an MFA factor in the Okta Web UI first!"
+                    "MFA not enrolled. Cannot continue. Please enroll an MFA factor in the Okta Web UI first!"
                 )
                 sys.exit(2)
             if status == "LOCKED_OUT":
@@ -134,9 +123,7 @@ def verify_mfa(self, factors_list, state_token):
                 if self.factor:
                     if self.factor == factor_provider:
                         factor_choice = index
-                        self.logger.info(
-                            "Using pre-selected factor choice from ~/.okta-aws"
-                        )
+                        self.logger.info("Using pre-selected factor choice from ~/.okta-aws")
                         break
                 else:
                     print("%d: %s" % (index + 1, factor_name))
@@ -146,12 +133,9 @@ def verify_mfa(self, factors_list, state_token):
                     self.okta_profile, supported_factors[factor_choice]["provider"]
                 )
             self.logger.info(
-                "Performing secondary authentication using: %s"
-                % supported_factors[factor_choice]["provider"]
-            )
-            session_token = self.verify_single_factor(
-                supported_factors[factor_choice], state_token
+                "Performing secondary authentication using: %s" % supported_factors[factor_choice]["provider"]
             )
+            session_token = self.verify_single_factor(supported_factors[factor_choice], state_token)
         else:
             print("MFA required, but no supported factors enrolled! Exiting.")
             exit(1)
@@ -178,9 +162,7 @@ def verify_single_factor(self, factor, state_token):
             elif resp_json["status"] == "MFA_CHALLENGE":
                 print("Waiting for push verification...")
                 while True:
-                    resp = requests.post(
-                        resp_json["_links"]["next"]["href"], json=req_data
-                    )
+                    resp = requests.post(resp_json["_links"]["next"]["href"], json=req_data)
                     resp_json = resp.json()
                     if resp_json["status"] == "SUCCESS":
                         return resp_json["sessionToken"]
@@ -204,9 +186,7 @@ def get_session(self, session_token):
         """Gets a session cookie from a session token"""
         data = {"sessionToken": session_token}
         # https://developer.okta.com/docs/guides/ie-limitations/main/#sessions-apis
-        resp = self._okta_json_request(
-            "POST", "/api/v1/sessions", "get_session", json=data
-        )
+        resp = self._okta_json_request("POST", "/api/v1/sessions", "get_session", json=data)
         self.cache_session_id(resp["id"], resp["expiresAt"])
         return resp["id"]
 
@@ -236,9 +216,7 @@ def get_cached_session_id(self):
 
         expiration_date = datetime.min
         if session_info.get("expiration_date"):
-            expiration_date = datetime.strptime(
-                session_info.get("expiration_date"), "%Y-%m-%dT%H:%M:%S.%fZ"
-            )
+            expiration_date = datetime.strptime(session_info.get("expiration_date"), "%Y-%m-%dT%H:%M:%S.%fZ")
 
         current_time = datetime.utcnow()
         if max([current_time, expiration_date]) == expiration_date:
@@ -252,17 +230,11 @@ def check_for_desync(self, session_id):
             sid = "sid=%s" % session_id
             headers = {"Cookie": sid}
             # https://developer.okta.com/docs/api/openapi/okta-management/management/tag/User/#tag/User/operation/getUser
-            raw_resp = requests.get(
-                self.https_base_url + "/api/v1/users/me", headers=headers
-            )
+            raw_resp = requests.get(self.https_base_url + "/api/v1/users/me", headers=headers)
             raw_resp.raise_for_status()
             return False
         except requests.HTTPError as e:
-            if (
-                e.response is None
-                or e.response.status_code != 403
-                or "Invalid session" not in e.response.text
-            ):
+            if e.response is None or e.response.status_code != 403 or "Invalid session" not in e.response.text:
                 raise e
             message = "Okta session invalidated. Refreshing token now..."
             self.logger.error(message)
@@ -321,9 +293,7 @@ def get_apps(self, session_id):
         sid = "sid=%s" % session_id
         headers = {"Cookie": sid}
         # https://developer.okta.com/docs/api/openapi/okta-management/management/tag/UserResources/#tag/UserResources/operation/listAppLinks
-        resp = self._okta_json_request(
-            "GET", "/api/v1/users/me/appLinks", "get_apps", headers=headers
-        )
+        resp = self._okta_json_request("GET", "/api/v1/users/me/appLinks", "get_apps", headers=headers)
 
         aws_apps = []
         for app in resp:
@@ -342,9 +312,7 @@ def get_apps(self, session_id):
             print("%d: %s" % (index + 1, app["label"]))
         if app_choice is None:
             app_choice = int(input("Please select AWS app: ")) - 1
-            self.okta_auth_config.save_chosen_app_for_profile(
-                self.okta_profile, aws_apps[app_choice]["label"]
-            )
+            self.okta_auth_config.save_chosen_app_for_profile(self.okta_profile, aws_apps[app_choice]["label"])
 
         return aws_apps[app_choice]["label"], aws_apps[app_choice]["linkUrl"]
 

oktaawscli/okta_auth_config.py

@@ -88,9 +88,7 @@ def region_for(self, okta_profile, default="us-east-1"):
 
     def get_check_valid_creds(self, okta_profile):
         """Gets if should check if AWS creds are valid from config"""
-        check_valid_creds = self._value.get(
-            okta_profile, "check-valid-creds", fallback="True"
-        )
+        check_valid_creds = self._value.get(okta_profile, "check-valid-creds", fallback="True")
         self.logger.info("Check if credentials are valid: %s" % check_valid_creds)
         return check_valid_creds
 
@@ -102,25 +100,17 @@ def get_store_role(self, okta_profile):
 
     def get_auto_write_profile(self, okta_profile):
         """Gets if should auto write aws creds to ~/.aws/credentials from config"""
-        auto_write_profile = self._value.get(
-            okta_profile, "auto-write-profile", fallback=True
-        )
-        self.logger.info(
-            "Should write profile to ~/.aws/credentials: %s" % auto_write_profile
-        )
+        auto_write_profile = self._value.get(okta_profile, "auto-write-profile", fallback=True)
+        self.logger.info("Should write profile to ~/.aws/credentials: %s" % auto_write_profile)
         return auto_write_profile
 
     def get_session_duration(self, okta_profile):
         """Gets STS session duration from config as an int"""
         # AWS docs say default duration is 1 hour (3600 seconds)
-        session_duration = int(
-            self._value.get(okta_profile, "session-duration", fallback="3600")
-        )
+        session_duration = int(self._value.get(okta_profile, "session-duration", fallback="3600"))
 
         if session_duration > 43200 or session_duration < 3600:
-            self.logger.info(
-                "Invalid session duration specified, defaulting to 1 hour."
-            )
+            self.logger.info("Invalid session duration specified, defaulting to 1 hour.")
             session_duration = 3600
 
         self.logger.info("Configured session duration: %s seconds" % session_duration)