Initial commit

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+venv

Dockerfile

@@ -0,0 +1,12 @@
+FROM continuumio/miniconda3
+
+RUN apt update
+RUN apt install git
+
+COPY entrypoint.sh /entrypoint.sh
+COPY *.py /
+COPY requirements.txt /
+
+RUN pip install -r requirements.txt
+
+ENTRYPOINT ["/entrypoint.sh"]

README.md

@@ -1,21 +1,22 @@
-# Garden Template
-The Garden Template contains sample files you could use for creating new [Code Garden](https://github.com/strongdm/garden) Repositories. It includes templates for:
-* [README](README-sample.md)
-* [License](LICENSE)
-* [Contributing](CONTRIBUTING.md)
-* [Support](SUPPORT.md)
-* Report [bug](.github/ISSUE_TEMPLATE/bug_report.md) or [feature requests](.github/ISSUE_TEMPLATE/feature_request.md)
-* [Pull Request](.github/PULL_REQUEST_TEMPLATE/pull_request_template.md)
-* [Documentation](docs)
+# SDM Access Github Action
 
-In order to use this repository, you could:
-* Use it as a Template - Green button at the top of the repo
-* Clone it and manually adjust it - Useful if you want to start a fresh project history
+Manage access to strongDM resources via Github Actions.
 
-After cloning the repo, remember to: 
-1. Remove this README file
-2. Rename the file README-sample.md to README.md and adjust the content
-3. Adjust the Contributing and Support guidelines
-4. Adjust the templates for bugs and feature requests under the .github folder 
+## Table of Contents
+* [Installation](#installation)
+* [Getting Started](#getting-started)
+* [Contributing](#contributing)
+* [Support](#support)
+
+## Installation
+Explain how to install the project/tool. Provide commands or animated GIFs if needed.
+
+## Getting Started
+Explain how to get quickly started with the tool. Provide commands or animated GIFs if needed, and create as many subsections as needed.
+
+## Contributing
+Refer to the [contributing](CONTRIBUTING.md) guidelines or dump part of the information here.
+
+## Support
+Refer to the [support](SUPPORT.md) guidelines or dump part of the information here.
 
-A template repo that can be used as a reference: [Auth0 Open Source Template](https://github.com/auth0/open-source-template)

action.yml

@@ -0,0 +1,5 @@
+name: 'SDM Access GH Action'
+description: 'Manage access to strongDM resources via Github Actions'
+runs:
+  using: 'docker'
+  image: 'Dockerfile'

entrypoint.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+
+echo "Checking pending requests..."
+git checkout main
+git pull
+git diff --name-only HEAD HEAD^ | while read line; do 
+    resource_name=$(echo $line | awk '/^resources/ { gsub("resources/", "", $0); print $0 }')
+    if [ "$resource_name" != "" ]; then
+        raw_user_email=$(git log -p -- $line | grep Author | head -1)
+        user_email=$(echo $raw_user_email | awk 'match($0, /<.*>/) { print substr($0, RSTART+1, RLENGTH-2) }')
+        echo "About to grant temporary access to $user_email on $resource_name"
+        python3 main.py $resource_name $user_email
+    fi
+done

main.py

@@ -0,0 +1,47 @@
+import datetime
+import logging
+import os
+import sdm_service
+import sys
+
+GRANT_TIMEOUT=60 #minutes
+
+def get_params():
+    if not sys.argv or len(sys.argv) != 3:
+        raise Exception("Invalid number of arguments")
+    return sys.argv[1], sys.argv[2]
+
+class GrantTemporaryAccess:
+    service = sdm_service.create_sdm_service(os.getenv("SDM_API_ACCESS_KEY"), os.getenv("SDM_API_SECRET_KEY"), logging)
+
+    def __init__(self, resource_name, user_email):
+        self.resource_name = resource_name
+        self.user_email = user_email
+
+    def __get_resource_id(self):
+        try: 
+            resource = self.service.get_resource_by_name(self.resource_name)
+            return resource.id
+        except Exception as e:
+            raise Exception(f"Invalid resource name {self.resource_name}") from e
+
+    def __get_account_id(self):
+        try: 
+            account = self.service.get_account_by_email(self.user_email)
+            return account.id
+        except Exception as e:
+            raise Exception(f"Invalid user email {self.user_email}") from e
+
+    def execute(self):
+        grant_start_from = datetime.datetime.now(datetime.timezone.utc)
+        grant_valid_until = grant_start_from + datetime.timedelta(minutes=GRANT_TIMEOUT)
+        self.service.grant_temporary_access(
+            self.__get_resource_id(), 
+            self.__get_account_id(), 
+            grant_start_from, 
+            grant_valid_until
+        )
+
+resource_name, user_email = get_params()
+GrantTemporaryAccess(resource_name, user_email).execute()
+print(f"Temporary grant successfullly created for {user_email} on {resource_name}")

requirements.txt

@@ -0,0 +1,2 @@
+strongdm
+

sdm_service.py

@@ -0,0 +1,140 @@
+# Copied from: https://github.com/strongdm/accessbot/blob/main/plugins/sdm/lib/service/sdm_service.py
+import strongdm
+
+def create_sdm_service(api_access_key, api_secret_key, log):
+    client = strongdm.Client(api_access_key, api_secret_key)
+    return SdmService(client, log)
+
+
+class NotFoundException(Exception):
+    pass
+
+class SdmService:
+    def __init__(self, client, log):
+        self.__client = client
+        self.__log = log
+
+    def get_resource_by_name(self, name):
+        """
+        Return a SDM resouce by name
+        """
+        try:
+            self.__log.debug("##SDM## SdmService.get_resource_by_name name: %s", name)
+            sdm_resources = list(self.__client.resources.list('name:"{}"'.format(name)))
+        except Exception as ex:
+            raise Exception("List resources failed: " + str(ex)) from ex
+        if len(sdm_resources) == 0:
+            raise NotFoundException("Sorry, cannot find that resource!")
+        return sdm_resources[0]
+
+    def get_account_by_email(self, email):
+        """
+        Return a SDM account by email
+        """
+        try:
+            self.__log.debug("##SDM## SdmService.get_account_by_email email: %s", email)
+            sdm_accounts = list(self.__client.accounts.list('email:{}'.format(email)))
+        except Exception as ex:
+            raise Exception("List accounts failed: " + str(ex)) from ex
+        if len(sdm_accounts) == 0:
+            raise Exception("Sorry, cannot find your account!")
+        return sdm_accounts[0]
+
+    def account_grant_exists(self, resource_id, account_id):
+        """
+        Does an account grant exists - resource assigned to an account
+        """
+        try:
+            self.__log.debug("##SDM## SdmService.account_grant_exists resource_id: %s account_id: %s", resource_id, account_id)
+            account_grants = list(self.__client.account_grants.list(f"resource_id:{resource_id},account_id:{account_id}"))
+            return len(account_grants) > 0
+        except Exception as ex:
+            raise Exception("Account grant exists failed: " + str(ex)) from ex
+
+    def role_grant_exists(self, resource_id, account_id):
+        """
+        Does a role grant exists - resource assigned to a role that is assigned to an account
+
+        account -> account_attachment -> role -> role_grant -> resource
+        """
+        try:
+            self.__log.debug("##SDM## SdmService.role_grant_exists resource_id: %s account_id: %s", resource_id, account_id)
+            for aa in list(self.__client.account_attachments.list(f"account_id:{account_id}")):
+                role = self.__client.roles.get(aa.role_id).role
+                for rg in list(self.__client.role_grants.list(f"role_id:{role.id}")):
+                    if rg.resource_id == resource_id:
+                        return True
+            return False
+        except Exception as ex:
+            raise Exception("Role grant exists failed: " + str(ex)) from ex
+
+    def grant_temporary_access(self, resource_id, account_id, start_from, valid_until):
+        """
+        Grant temporary access to a SDM resource for an account
+        """
+        try:
+            self.__log.debug(
+                "##SDM## SdmService.grant_temporary_access resource_id: %s account_id: %s start_from: %s valid_until: %s",
+                resource_id, account_id, str(start_from), str(valid_until)
+            )
+            sdm_grant = strongdm.AccountGrant(
+                resource_id = resource_id,
+                account_id = account_id,
+                start_from = start_from,
+                valid_until = valid_until
+            )
+            self.__client.account_grants.create(sdm_grant)
+        except Exception as ex:
+            raise Exception("Grant failed: " + str(ex)) from ex
+
+    def get_all_resources(self, filter = ''):
+        """
+        Return all resources
+        """
+        self.__log.debug("##SDM## SdmService.get_all_resources")
+        try:
+            return self.remove_none_values(self.__client.resources.list(filter))
+        except Exception as ex:
+            raise Exception("List resources failed: " + str(ex)) from ex
+
+    def get_all_resources_by_role(self, role_name, filter = ''):
+        """
+        Return all resources by role name
+        """
+        self.__log.debug("##SDM## SdmService.get_all_resources_by_role_name role_name: %s", role_name)
+        try:
+            sdm_role = self.get_role_by_name(role_name)
+            sdm_role_grants = list(self.__client.role_grants.list(f"role_id:{sdm_role.id}"))
+            resources_filter = ",".join([f"id:{rg.resource_id}" for rg in sdm_role_grants])
+            if filter:
+                resources_filter += f",{filter}"
+            return self.remove_none_values(self.__client.resources.list(resources_filter))
+        except Exception as ex:
+            raise Exception("List resources by role failed: " + str(ex)) from ex
+
+    def get_role_by_name(self, name):
+        """
+        Return a SDM role by name
+        """
+        try:
+            self.__log.debug("##SDM## SdmService.get_role_by_name name: %s", name)
+            sdm_roles = list(self.__client.roles.list('name:"{}"'.format(name)))
+        except Exception as ex:
+            raise Exception("List roles failed: " + str(ex)) from ex
+        if len(sdm_roles) == 0:
+            raise NotFoundException("Sorry, cannot find that role!")
+        return sdm_roles[0]
+
+    def get_all_roles(self):
+        """
+        Return all roles
+        """
+        self.__log.debug("##SDM## SdmService.get_all_roles")
+        try:
+            return list(self.__client.roles.list(''))
+        except Exception as ex:
+            raise Exception("List roles failed: " + str(ex)) from ex
+
+    @staticmethod
+    def remove_none_values(elements):
+        return [e for e in elements if e is not None]