feat: add --cdk-out

.github/workflows/ci.yml

@@ -14,21 +14,21 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os:
-        - 'ubuntu-latest'
-        # - 'windows-latest'
-        # - 'mac-latest'
+        os: ['ubuntu-latest']
+        python-version: ['3.10', '3.11', '3.12', '3.13']
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
 
       - uses: actions/setup-python@v5
         with:
-          python-version: '3.13'
+          python-version: ${{ matrix.python-version }}
           cache: 'pip'
 
       - name: install deps
-        run: pip install -r requirements.txt -r requirements-dev.txt
+        run: |
+          pip install -r requirements-dev.txt
+          pip install --editable .
 
       - name: lint
         run: ruff check
@@ -38,6 +38,3 @@ jobs:
 
       - name: test
         run: pytest --cov=./src/ ./tests/
-
-      - name: coverage
-        run: coverage report

.gitignore

@@ -178,3 +178,6 @@ volume
 
 # aws
 aws
+
+# make marker files
+requirements*.hash

Makefile

@@ -1,9 +1,41 @@
+SHELL := /bin/bash
+.SHELLFLAGS = -ec
+.DEFAULT_GOAL = help
+.PHONY = help clean format test install-deps
+
+REQ_FILES := requirements.txt requirements-dev.txt
+REQ_HASH := requirements.hash
+
 help:
 	@egrep -h '\s##\s' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m  %-30s\033[0m %s\n", $$1, $$2}'
 
-format:  ## format the project
+# Generate a hash file based on the last modified times of REQ_FILES
+# HANDLE IF ERROR TO CLEAN REQ_HASH
+$(REQ_HASH): $(REQ_FILES)
+	@echo "Checking if dependencies need to be installed..."
+	stat -c %Y $(REQ_FILES) | md5sum | awk '{print $$1}' > $(REQ_HASH)
+	@echo "Installing dependencies..."
+	python3 -m venv ./.venv
+	pip install -e . -r requirements-dev.txt
+
+install-deps: $(REQ_HASH)  ## Install dependencies if requirements files changed
+
+format:  ## Format the project
 	ruff format
 	ruff check --fix
 
-test:  ## run unit tests and generate coverage report
+validate:  ## Validate the project is linted and formatted
+	ruff format --check
+	ruff check
+
+test:  ## Run unit tests and generate coverage report
 	pytest --cov=./src/ ./tests/
+
+clean:  ## Clean generated project files
+	rm -f $(REQ_HASH)
+	rm -f .coverage
+	rm -rf ./.ruff_cache
+	rm -rf ./.pytest_cache
+	rm -rf ./.venv
+	rm -rf ./.tox
+	find . -type d -name "__pycache__" -exec rm -r {} +

README.md

@@ -6,11 +6,31 @@
 
 cycl is a CLI and Python SDK to help identify cross-stack import/export circular dependencies, for a given AWS account and region.
 
-### Why use cycl?
+## Getting started
 
-Over the lifetime of a project, circular references are bound to be introduced. They may not be noticed until you need to re-deploy some infrastructure. A good example is disaster recovery testing and potentially deploying all your infrastructure from scratch in a new region. This tool allows you to scan 
+Install `cycl` by running `pip install cycl`.
 
-## Getting Started
+### CLI
 
-The project is intended to be ran and published via a 
+- `cycl check --exit-zero` - exit 0 regardless of result
+- `cycl check --log-level` - set the logging level (default: WARNING)
+- `cycl check --cdk-out /path/to/cdk.out` - path to cdk.out, where stacks are CDK synthesized to CFN templates
 
+### SDK
+
+...
+
+## How to use cycl?
+
+There are two main use cases for `cycl`.
+
+1. In a pipeline. `cycl` is best used to detect circular dependencies before a deployment. If you're using the AWS CDK v2 (v1 support coming soon), simply synthesize you templates to a directory and pass that directory to `cycl` using `--cdk-out-path some-path-here `. This allows `cycl` to find all existing cycles and then those to be introduced by the deployment. This prevents the circular dependency from ever being introduced. If your pipeline deploys more than once, you should execute `cycl` before each deployment.
+2. To perform analysis. While a CLI is best used in a pipeline, if you require analysis which is not currently supported, you can use the SDK. The SDK gives you all the information that `cycl` collects.
+
+## Why use cycl?
+
+Over the lifetime of a project, circular references are bound to be introduced. They may not be noticed until you need to re-deploy some infrastructure. A good example is disaster recovery testing and potentially deploying all your infrastructure from scratch in a new region. This tool detects those changes.
+
+## Contributing
+
+`cycl` is being actively developed, instructions to come as it becomes more stable.

TODO.md

@@ -1,14 +1,20 @@
 # TODO.md
 
+- output the topological generations of the graph
+  - `cycl check --generations`
+- ignoring known cycles, [this](https://cs.stackexchange.com/questions/90481/how-to-remove-cycles-from-a-directed-graph) may help
+  - `cycl check --ignore-cycle-contains`
+  - `cycl check --ignore-cycle`
+- reducing the stacks, for example, a tag on a stack representing the github repo name
+  - `cycl check --reduce-dependencies-on`
+  - `cycl check --reduce-generations-on`
+
 - [ ] [fully configure dependabot](https://docs.github.com/en/code-security/dependabot/working-with-dependabot/automating-dependabot-with-github-actions)
 - [ ] configure make file
 - [ ] dprint for other file formatting
 - [ ] automatic release versioning for pypi, run unit tests against package? Is this possible?
-- [ ] isort
 - [ ] configuration file using cli, use toml, custom rc file, env vars?
-
-
-Next steps:
-1. Add `cycl --check` as first feature, publish v1.0.0 to pypi
-2. Add ability to ignore known cycles, how to break the cycle?
-3. Add ability to reduce via tags, check out jquery --reduce-on maybe?
+- [ ] CDK v1 support
+- [ ] Test with stages that the correct manifest is analyzed
+- [ ] What if cdk out synth is for multiple accounts? We may need to determine what account we have credentials for and only analyze those templates
+- [ ] automatic documentation generation

pyproject.toml

@@ -3,18 +3,30 @@ requires = ['setuptools>=64', 'setuptools-scm>=8']
 build-backend = 'setuptools.build_meta'
 
 [project]
+
 name = 'cycl'
-version='0.0.2'
+version='0.1.1'
+requires-python = ">=3.8"
 description = 'CLI and Python SDK to help identify cross-stack import/export circular dependencies, for a given AWS account and region.'
 readme = 'README.md'
 keywords = ['aws', 'cdk', 'cycle', 'circular', 'dependency', 'infrastructure']
 classifiers = [
     'Intended Audience :: Developers',
     'Natural Language :: English',
     'Programming Language :: Python :: 3',
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     'Programming Language :: Python',
     'Topic :: Software Development',
 ]
+dependencies = [
+    'boto3~=1.0',
+    'networkx~=3.0',
+]
 
 [project.urls]
 Repository = 'http://github.com/tcm5343/cycl'
@@ -69,3 +81,4 @@ omit = [
 
 [tool.pytest.ini_options]
 log_cli_level = "INFO"
+addopts = ['--import-mode=importlib']

src/cycl/cli.py

@@ -1,11 +1,15 @@
 import argparse
 import logging
+import pathlib
 import sys
+from logging import getLogger
 
 import networkx as nx
 
 from cycl import build_dependency_graph
-from cycl.utils.log_config import configure_logging
+from cycl.utils.log_config import configure_log
+
+log = getLogger(__name__)
 
 
 def app() -> None:
@@ -17,22 +21,28 @@ def app() -> None:
     parent_parser.add_argument(
         '--log-level',
         choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
-        default='ERROR',
-        help='Set the logging level (default: ERROR)',
+        default='WARNING',
+        help='set the logging level (default: WARNING)',
+    )
+    parent_parser.add_argument(
+        '--cdk-out',
+        type=pathlib.Path,
+        help='path to cdk.out, where stacks are CDK synthesized to CFN templates',
     )
 
     subparsers.add_parser('check', parents=[parent_parser], help='Check for cycles in AWS stack imports/exports')
 
     args = parser.parse_args()
-    configure_logging(getattr(logging, args.log_level))
+    configure_log(getattr(logging, args.log_level))
+    log.info(args)
 
     if args.action == 'check':
         cycle_found = False
-        graph = build_dependency_graph()
+        graph = build_dependency_graph(cdk_out_path=args.cdk_out)
         cycles = nx.simple_cycles(graph)
         for cycle in cycles:
             cycle_found = True
-            print(f'Cycle found between nodes: {cycle}')
+            print(f'cycle found between nodes: {cycle}')
         if cycle_found and not args.exit_zero:
             sys.exit(1)
     sys.exit(0)

src/cycl/cycl.py

@@ -1,19 +1,37 @@
+from __future__ import annotations
+
 from logging import getLogger
+from pathlib import Path
 
 import networkx as nx
 
+from cycl.utils.cdk import get_cdk_out_imports
 from cycl.utils.cfn import get_all_exports, get_all_imports, parse_name_from_id
 
 log = getLogger(__name__)
 
 
-def build_dependency_graph() -> nx.MultiDiGraph:
+def build_dependency_graph(cdk_out_path: Path | None = None) -> nx.MultiDiGraph:
     dep_graph = nx.MultiDiGraph()
 
+    cdk_out_imports = {}
+    if cdk_out_path:
+        cdk_out_imports = get_cdk_out_imports(Path(cdk_out_path))
+
     exports = get_all_exports()
+    # this could be made more efficient if get_all_exports returns a dict instead of a list, no need to iterate through
+    for export_name in cdk_out_imports:
+        if not any(export_name == export['Name'] for export in exports):
+            log.warning(
+                'found an export (%s) which has not been deployed yet about to be imported stack(s): (%s)',
+                export_name,
+                cdk_out_imports[export_name],
+            )
+
     for export in exports:
         export['ExportingStackName'] = parse_name_from_id(export['ExportingStackId'])
         export['ImportingStackNames'] = get_all_imports(export_name=export['Name'])
+        export.setdefault('ImportingStackNames', []).extend(cdk_out_imports.get(export['Name'], []))
         edges = [
             (export['ExportingStackName'], importing_stack_name) for importing_stack_name in export['ImportingStackNames']
         ]

src/cycl/utils/cdk.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import json
+from logging import getLogger
+from os import walk
+from pathlib import Path
+
+log = getLogger(__name__)
+
+
+class InvalidCdkOutPathError(Exception):
+    def __init__(self, message: str = 'An error occurred') -> None:
+        super().__init__(message)
+
+
+def __find_import_values(data: dict) -> list[str]:
+    """recursively search for all 'Fn::ImportValue' keys and return their values"""
+    results = []
+
+    if isinstance(data, dict):
+        for key, value in data.items():
+            if key == 'Fn::ImportValue':
+                results.append(value)
+            else:
+                results.extend(__find_import_values(value))
+
+    elif isinstance(data, list):
+        for item in data:
+            results.extend(__find_import_values(item))
+
+    return results
+
+
+def __get_import_values_from_template(file_path: Path) -> list[str]:
+    """todo: handle yaml templates too"""
+    with Path.open(file_path) as f:
+        json_data = json.load(f)
+    return __find_import_values(json_data)
+
+
+def __get_stack_name_from_manifest(path_to_manifest: Path, template_file_name: str) -> str:
+    with Path.open(path_to_manifest) as f:
+        json_data = json.load(f)
+    return json_data['artifacts'][template_file_name.split('.')[0]]['displayName']
+
+
+def __validate_cdk_out_path(cdk_out_path: Path) -> Path:
+    errors = []
+    if Path.exists(cdk_out_path):
+        if not cdk_out_path.is_dir():
+            errors.append('path must be a directory')
+    else:
+        errors.append("path doesn't exist")
+
+    # handle if path is where cdk.out/ is or is directly to cdk.out
+    if Path(cdk_out_path).name != 'cdk.out':
+        cdk_out_path = Path(cdk_out_path) / 'cdk.out'
+
+    if not Path.exists(Path(cdk_out_path) / 'cdk.out'):
+        errors.append('unable to find CDK stack synthesis output in provided directory, did you synth?')
+
+    if errors:
+        errors_formatted = '\n\t - '.join(errors)
+        error_message = f'Invalid path provided for --cdk-out {cdk_out_path}:\n\t - {errors_formatted}'
+        raise InvalidCdkOutPathError(error_message)
+
+    return cdk_out_path
+
+
+def get_cdk_out_imports(cdk_out_path: Path) -> dict[str, list[str]]:
+    """
+    map an export name to a list of stacks which import it
+
+    function does not take into consideration exports
+        - if we found an export, we may not be able to resolve the name of it
+        - AWS has built in circular dependency detection inside of a stack
+        - a circular dependency couldn't be introduced in a single deployment
+    """
+    cdk_out_path = __validate_cdk_out_path(cdk_out_path)
+
+    stack_import_mapping = {}
+    for root, _dirs, files in walk(cdk_out_path):
+        for file in files:
+            if file.endswith('template.json'):
+                imported_export_names = __get_import_values_from_template(Path(root) / file)
+                if imported_export_names:
+                    manifest_path = Path(root) / 'manifest.json'
+                    stack_name = __get_stack_name_from_manifest(manifest_path, file)
+                    for export_name in imported_export_names:
+                        stack_import_mapping.setdefault(export_name, []).append(stack_name)
+    return stack_import_mapping

src/cycl/utils/cfn.py

@@ -26,10 +26,13 @@ def get_all_exports(cfn_client: BaseClient | None = None) -> list[dict]:
 
     exports = []
     resp = cfn_client.list_exports()
+    log.debug(resp)
     exports.extend(resp['Exports'])
     while token := resp.get('NextToken'):
         resp = cfn_client.list_exports(NextToken=token)
+        log.debug(resp)
         exports.extend(resp['Exports'])
+    log.debug(exports)
     return exports
 
 
@@ -40,12 +43,15 @@ def get_all_imports(export_name: str, cfn_client: BaseClient | None = None) -> l
     imports = []
     try:
         resp = cfn_client.list_imports(ExportName=export_name)
+        log.debug(resp)
         imports.extend(resp['Imports'])
         while token := resp.get('NextToken'):
             resp = cfn_client.list_imports(ExportName=export_name, NextToken=token)
+            log.debug(resp)
             imports.extend(resp['Imports'])
     except ClientError as err:
         if 'is not imported by any stack' not in repr(err):
             raise
         log.debug('')
+    log.debug(imports)
     return imports