v0.2.0: bug fixes + auto-publish workflow (#50)

* v0.2.0: bug fixes + auto-publish workflow

* stub tests

* stub for tests

.github/workflows/ci.yml

@@ -1,16 +1,10 @@
-name: 'CI: Pytest + Coverage'
+name: 'CI: Pytest, Codecov and PyPI Publish'
 
 on:
   # manual trigger
   workflow_dispatch:
-  push:
-    branches:
-      - main
-    paths:
-      - 'aws-sso-manager/**'
-      - 'tests/**'
-      - 'poetry.lock'
-      - 'pyproject.toml'
+
+  # pull requests
   pull_request:
     branches:
       - main
@@ -20,8 +14,13 @@ on:
       - 'poetry.lock'
       - 'pyproject.toml'
 
+  # release
+  release:
+    types: [published]
+
 jobs:
-  # Build
+
+  # Build, test and upload codecov report
   python-test:
     name: pytest
     runs-on: ${{ matrix.os }}
@@ -43,16 +42,36 @@ jobs:
           architecture: x64
           cache: poetry
 
-      - name: Install aws-sso-manager
+      - name: Install awsssomanager
         run: poetry install --no-interaction --no-ansi
 
-      # Commented out for template
-      #
-      # - name: Run Tests
-      #   run: poetry run pytest --cov aws-sso-manager
+      - name: Run Tests
+        run: poetry run pytest --cov awsssomanager
+
+      - name: Upload coverage reports to Codecov
+        uses: codecov/codecov-action@v3
+        env: 
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+        if: ${{ matrix.os == 'ubuntu-latest' && matrix.python == 3.12 }}
+
+  # Publish
+  publish:
+    needs: python-test
+    if: github.event_name == 'release'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.12
+          architecture: x64
+
+      - name: Install Poetry
+        run: pipx install poetry
 
-      # - name: Upload coverage reports to Codecov
-      #   uses: codecov/codecov-action@v3
-      #   env: 
-      #     CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-      #   if: ${{ matrix.os == 'ubuntu-latest' && matrix.python == 3.9 }}
+      - name: Build and publish
+        run: |
+          poetry publish --build --username __token__ --password ${{ secrets.PYPI_API_TOKEN }}

awsssomanager/config/config.py

@@ -23,6 +23,24 @@ class AWSSSOManagerConfig:
     def __init__(self, config: Dict):
         self.config: Dict = config
 
+    @property
+    def access_token(self) -> str:
+        """Get the accessToken
+
+        Returns:
+            accessToken.
+        """
+        return self.config['default']['accessToken']
+
+    @property
+    def access_token_expires_at(self) -> float:
+        """Get the accessTokenExpiresAt
+
+        Returns:
+            accessTokenExpiresAt.
+        """
+        return float(self.config['default'].get('accessTokenExpiresAt', '0'))
+
     @property
     def login_account(self) -> str:
         """Get the loginAccount
@@ -69,7 +87,7 @@ def from_config(cls, config_file: str):
         Returns:
             class instance.
         """
-        with open(config_file, 'r') as file:
+        with open(config_file, 'r', encoding='utf-8') as file:
             config = yaml.safe_load(file)
         return cls(config=config)
 
@@ -114,6 +132,6 @@ def check_aws_sso_config_exists() -> bool:
     def write_config(self) -> None:
         """Write aws-sso-manger config file
         """
-        with open(SSO_MANAGER_CONFIG_FILE, 'w') as file:
+        with open(SSO_MANAGER_CONFIG_FILE, 'w', encoding='utf-8') as file:
             logger.info(f'writing config file {SSO_MANAGER_CONFIG_FILE}')
             yaml.safe_dump(self.config, file, default_flow_style=False)

awsssomanager/sso/credentials.py

@@ -2,11 +2,11 @@
 from concurrent.futures import ThreadPoolExecutor
 import configparser
 import logging
+import sys
 import time
 from typing import Dict
 
 import boto3
-import botocore
 
 from awsssomanager import CREDENTIALS_FILE
 from awsssomanager.config import AWSSSOManagerConfig
@@ -28,26 +28,17 @@
 logger = logging.getLogger(__name__)
 
 
-def get_aws_sso_client() -> boto3.client:
-    """Get an AWS SSO client.
-
-    Returns:
-        boto3.client
-    """
-    return boto3.client("sso", region_name="us-east-1")
-
-
-def get_accounts_info(access_token: str) -> Dict:
+def get_accounts_info(config: AWSSSOManagerConfig) -> Dict:
     """Retrieve account information for all accounts that a user
     has access to.
 
     Args:
-        access_token (str): The access token for authentication.
+        config (AWSSSOManagerConfig): AWSSSOManagerConfig.
 
     Returns:
         Dict: A dictionary containing account information.
     """
-    sso_client = get_aws_sso_client()
+    sso_client = boto3.client("sso", region_name=config.region)
     accounts_info = {}
     list_accounts_paginator = sso_client.get_paginator("list_accounts")
 
@@ -60,7 +51,7 @@ def process_page(page):
                 "emailAddress": account["emailAddress"],
             }
 
-    for page in list_accounts_paginator.paginate(accessToken=access_token):
+    for page in list_accounts_paginator.paginate(accessToken=config.access_token):
         process_page(page)
 
     return accounts_info
@@ -76,38 +67,35 @@ def get_credentials(config: AWSSSOManagerConfig) -> None:
         None
     """
     credentials = load_credentials()
-    sso_client = boto3.client("sso-oidc", region_name="us-east-1")
-    token_expires_at = float(config.config["default"].get("accessTokenExpiresAt", "0"))
-    access_token = config.config["default"].get("accessToken", "")
+    sso_client = boto3.client("sso-oidc", region_name=config.region)
 
     logger.info('Retrieving credentials')
     while True:
         try:
-            if time.time() > token_expires_at:
+            if time.time() > config.access_token_expires_at:
                 # token expired, create a new one
                 try:
                     config = create_token(config)
                 except sso_client.exceptions.InvalidGrantException:
                     # invalid grant, reauthorize device
                     logger.warning("Unable to create token, reauthorizing...")
                     config = start_device_authorization(config)
-                access_token = config.config["default"]["accessToken"]
-                token_expires_at = float(config.config["default"]["accessTokenExpiresAt"])
+                except Exception as error:
+                    logger.error(f"Unable to create token, unknown exception: {error}")
+                    sys.exit(1)
 
             # Get list of accounts user has access to
-            accounts = get_accounts_info(access_token)
-            logger.debug(accounts)
+            accounts = get_accounts_info(config=config)
 
             # Get list of roles (for each account) the user has access to
-            roles = get_roles_for_accounts(accounts.values(), access_token)
-            logger.debug(roles)
+            roles = get_roles_for_accounts(accounts.values(), config)
 
             # Get credentials for all roles
             new_credentials = {}
             with ThreadPoolExecutor(max_workers=10) as pool:
                 results = list(pool.map(
                     lambda role: get_role_credentials(
-                        role, access_token, accounts
+                        accounts_info=accounts, config=config, role=role
                     ),
                     roles,
                 ))
@@ -116,7 +104,6 @@ def get_credentials(config: AWSSSOManagerConfig) -> None:
             for result in results:
                 new_credentials.update(result)
 
-            logger.debug(new_credentials)
             # Iterate over roles for account default permissions
             with ThreadPoolExecutor(max_workers=10) as pool:
                 pool.map(
@@ -137,7 +124,7 @@ def get_credentials(config: AWSSSOManagerConfig) -> None:
             logger.info("Credentials successfully retrieved")
             break  # exit out of cred loop
 
-        except sso_client.exceptions.UnauthorizedException:
+        except sso_client.exceptions.UnauthorizedClientException:
             logger.info("Access token unauthorized, sleeping and refetching credentials...")
             time.sleep(30)
 
@@ -163,7 +150,8 @@ def update_credentials_for_role(
     profile_name = f"{account_id}_{role_name}"
     account_name = accounts_info[account_id]["accountName"]
 
-    # If the account id alias is already in credentials, check to see if we should override for higher priority
+    # If the account id alias is already in credentials,
+    # check to see if we should override for higher priority
     if account_id in new_credentials:
         if compare_roles(config, role_name, new_credentials[account_id]["aws_sso_role_name"]) > 0:
             new_credentials[account_id] = new_credentials[profile_name].copy()
@@ -197,6 +185,6 @@ def save_credentials(credentials: configparser.ConfigParser) -> configparser.Con
     Returns:
         configparser.ConfigParser: ConfigParser object after saving.
     """
-    with open(CREDENTIALS_FILE, "w") as credentialsfile:
-        credentials.write(credentialsfile)
+    with open(CREDENTIALS_FILE, "w", encoding="utf-8") as file:
+        credentials.write(file)
     return load_credentials()

awsssomanager/sso/device.py

@@ -54,9 +54,9 @@ def register_new_device(config: AWSSSOManagerConfig) -> AWSSSOManagerConfig:
     Returns:
         AWSSSOMangerConfig instance.
     """
-    sso_client = boto3.client("sso-oidc", region_name="us-east-1")
+    sso_client = boto3.client("sso-oidc", region_name=config.region)
     register_results = sso_client.register_client(
-        clientName="awsssomanager", clientType="public"
+        clientName="aws-sso-manager", clientType="public"
     )
     for key in ["clientId", "clientSecret", "clientSecretExpiresAt"]:
         config.config["default"][key] = str(register_results[key])
@@ -89,6 +89,7 @@ def start_device_authorization(
         clientSecret=config.config["default"]["clientSecret"],
         startUrl=f"https://{config.sso_domain}.awsapps.com/start",
     )
+
     device_auth_starttime = time.time()
     config.config["default"]["deviceCode"] = device_auth_results["deviceCode"]
     verification_uri = device_auth_results["verificationUriComplete"]
@@ -105,5 +106,6 @@ def start_device_authorization(
         time.sleep(1)
         if (time.time() - device_auth_starttime) >= device_auth_results["expiresIn"]:
             raise RuntimeError("Failed to authenticate in time")
+
     logger.info('Successfully registered user.')
     return config

awsssomanager/sso/roles.py

@@ -7,6 +7,7 @@
 
 from awsssomanager.config.config import AWSSSOManagerConfig
 
+
 __all__ = [
     "compare_roles",
     "get_roles_for_accounts",
@@ -50,33 +51,33 @@ def compare_roles(
     return 0
 
 
-def get_roles_for_accounts(accounts: List[Dict], access_token: str) -> List[Dict]:
+def get_roles_for_accounts(accounts: List[Dict], config: AWSSSOManagerConfig) -> List[Dict]:
     """Retrieve roles for a list of accounts.
 
     Args:
         accounts (List[Dict]): List of accounts.
-        access_token (str): The access token for authentication.
+        config (AWSSSOManagerConfig): AWSSSOManagerConfig.
 
     Returns:
         List[Dict]: List of roles.
     """
-    sso_client = boto3.client("sso", region_name="us-east-1")
+    sso_client = boto3.client("sso", region_name=config.region)
     roles = []
     list_roles_paginator = sso_client.get_paginator("list_account_roles")
 
     for account in accounts:
         page = list_roles_paginator.paginate(
-            accessToken=access_token, accountId=account["accountId"]
+            accessToken=config.access_token, accountId=account["accountId"]
         ).build_full_result()
         roles.extend(page["roleList"])
 
     return roles
 
 
 def get_role_credentials(
-    role: Dict,
-    access_token: str,
-    accounts_info: Dict
+    accounts_info: Dict,
+    config: AWSSSOManagerConfig,
+    role: Dict
 ) -> Dict:
     """Retrieve credentials for a role.
 
@@ -88,11 +89,11 @@ def get_role_credentials(
     Returns:
         Dict: A dictionary containing role credentials.
     """
-    sso_client = boto3.client("sso", region_name="us-east-1")
+    sso_client = boto3.client("sso", region_name=config.region)
     account_id = role["accountId"]
 
     role_credentials = sso_client.get_role_credentials(
-        accessToken=access_token, accountId=account_id, roleName=role['roleName']
+        accessToken=config.access_token, accountId=account_id, roleName=role['roleName']
     )["roleCredentials"]
 
     return {

awsssomanager/sso/token.py

@@ -24,7 +24,7 @@ def create_token(config: AWSSSOManagerConfig) -> AWSSSOManagerConfig:
     Returns:
         AWSSSOManagerConfig: Updated configuration after creating the token.
     """
-    sso_client = boto3.client("sso-oidc", region_name="us-east-1")
+    sso_client = boto3.client("sso-oidc", region_name=config.region)
     token_result = sso_client.create_token(
         clientId=config.config["default"]["clientId"],
         clientSecret=config.config["default"]["clientSecret"],
@@ -51,7 +51,7 @@ def test_create_token(config: AWSSSOManagerConfig) -> bool:
     Returns:
         bool: True if successful else False.
     """
-    sso_client = boto3.client("sso-oidc", region_name="us-east-1")
+    sso_client = boto3.client("sso-oidc", region_name=config.region)
     try:
         create_token(config)
     except (

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "awsssomanager"
-version = "0.1.3"
+version = "0.2.0"
 description = "AWS SSO Manager"
 authors = ["Brett Fitzpatrick"]
 license = "MIT"

tests/awsssomanager/config/test_config.py

@@ -0,0 +1,21 @@
+"""awsssomanager.config.Config tests."""
+
+from awsssomanager.config import AWSSSOManagerConfig
+
+
+def test_create_aws_dir():
+    """Test create_aws_dir."""
+    AWSSSOManagerConfig.create_aws_dir()
+    assert AWSSSOManagerConfig.check_aws_dir_exists()
+
+
+def test_from_config(tmp_path):
+    """Test from_config."""
+    config_file = tmp_path / "config.yaml"
+    config_file.write_text(
+        "default:\n  accessToken: test\n  loginAccount: test\n  region: us-east-1"
+    )
+    config = AWSSSOManagerConfig.from_config(config_file=config_file)
+    assert config.access_token == "test"
+    assert config.login_account == "test"
+    assert config.region == "us-east-1"