GitHub Enterprise/Cloud support

README.md

@@ -18,9 +18,9 @@ Please refer to the [Releases](https://github.com/checkmarx-ts/cxone-flow/releas
 * Supported SCMs
     * BitBucket Data Center
     * Azure DevOps Enterprise
+    * GitHub Enterprise and Cloud
 * Scans are invoked by Push events when code is pushed to protected branches.
 * Scans are invoked on Pull-Requests that target a protected branch.
 * Scan results for Pull-Request scans are summarized in a pull-request comment.
-* Pull-Request state is reflected in scan tags as the pull request is under
-review.
+* Pull-Request state is reflected in scan tags as the pull request is under review.
 

api_utils/__init__.py

@@ -1,23 +1,15 @@
-from .apisession import APISession
-from requests.auth import AuthBase,HTTPBasicAuth
 from .signatures import signature
+from .auth_factories import AuthFactory, StaticAuthFactory
+from requests.auth import HTTPBasicAuth
+from .bearer import HTTPBearerAuth
+import urllib
 
 
-def auth_basic(username, password) -> AuthBase:
-    return HTTPBasicAuth(username, password)
+def auth_basic(username, password) -> AuthFactory:
+    return StaticAuthFactory(HTTPBasicAuth(username, password))
 
-
-def auth_bearer(token) -> AuthBase:
-    class HTTPBearerAuth(AuthBase):
-        def __init__(self, token):
-            AuthBase.__init__(self)
-            self.__token = token
-
-        def __call__(self, r):
-            r.headers["Authorization"] = f"Bearer {self.__token}"
-            return r
-
-    return HTTPBearerAuth(token)
+def auth_bearer(token) -> AuthFactory:
+    return StaticAuthFactory(HTTPBearerAuth(token))
 
 def verify_signature(signature_header, secret, body) -> bool:
     (algorithm, hash) = signature_header.split("=")
@@ -31,4 +23,9 @@ def verify_signature(signature_header, secret, body) -> bool:
 
     return generated_hash == hash
 
+def form_url(endpoint : str, url_path : str, anchor=None, **kwargs):
+    base = endpoint.rstrip("/")
+    suffix = urllib.parse.quote(url_path.lstrip("/"))
+    args = [f"{x}={urllib.parse.quote(str(kwargs[x]))}" for x in kwargs.keys()]
+    return f"{base}/{suffix}{"?" if len(args) > 0 else ""}{"&".join(args)}{f"#{anchor}" if anchor is not None else ""}"
 

api_utils/apisession.py

@@ -1,10 +1,11 @@
 from _agent import __agent__
-from requests.auth import AuthBase
 from requests import Response
 from requests import request
 from typing import Dict, Union, Any
-import urllib, logging, sys, asyncio
-
+import logging, sys, asyncio
+from api_utils import AuthFactory
+from api_utils.auth_factories import EventContext
+from . import form_url
 
 class SCMAuthException(Exception):
     pass
@@ -18,28 +19,31 @@ class APISession:
     def log(clazz):
         return logging.getLogger(clazz.__name__)
 
-    def __init__(self, api_base_endpoint : str, auth : AuthBase, timeout : int = 60, retries : int = 3, proxies : Dict = None, ssl_verify : Union[bool, str] = True):
+    def __init__(self, api_endpoint : str, auth : AuthFactory, timeout : int = 60, retries : int = 3, proxies : Dict = None, ssl_verify : Union[bool, str] = True):
 
         self.__headers = { "User-Agent" : __agent__ }
 
-        self.__base_endpoint = api_base_endpoint
+        self.__api_endpoint = api_endpoint
         self.__timeout = timeout
         self.__retries = retries
 
         self.__verify = ssl_verify
         self.__proxies = proxies
-        self.__auth = auth
-
-
-    def _form_url(self, url_path, anchor=None, **kwargs):
-        base = self.__base_endpoint.rstrip("/")
-        suffix = urllib.parse.quote(url_path.lstrip("/"))
-        args = [f"{x}={urllib.parse.quote(str(kwargs[x]))}" for x in kwargs.keys()]
-        return f"{base}/{suffix}{"?" if len(args) > 0 else ""}{"&".join(args)}{f"#{anchor}" if anchor is not None else ""}"
-
-
-    async def exec(self, method : str, path : str, query : Dict = None, body : Any = None, extra_headers : Dict = None) -> Response:
-        url = self._form_url(path)
+        self.__auth_factory = auth
+
+    @staticmethod
+    def form_api_endpoint(base_endpoint : str, suffix : str):
+        ret = base_endpoint.rstrip("/")
+        if suffix is not None and len(suffix) > 0:
+            ret = f"{ret}/{suffix.lstrip("/").rstrip("/")}"
+        return ret
+
+    @property
+    def api_endpoint(self):
+        return self.__api_endpoint
+
+    async def exec(self, event_context : EventContext, method : str, path : str, query : Dict = None, body : Any = None, extra_headers : Dict = None) -> Response:
+        url = form_url(self.api_endpoint, path)
         headers = dict(self.__headers)
         if not extra_headers is None:
             headers.update(extra_headers)
@@ -50,8 +54,8 @@ async def exec(self, method : str, path : str, query : Dict = None, body : Any =
 
             APISession.log().debug(f"Executing: {prepStr} #{tryCount}")
             response = await asyncio.to_thread(request, method=method, url=url, params=query,
-                data=body, headers=headers, auth=self.__auth, timeout=self.__timeout, 
-                proxies=self.__proxies, verify=self.__verify)
+                data=body, headers=headers, auth=await self.__auth_factory.get_auth(event_context, tryCount > 0), 
+                timeout=self.__timeout, proxies=self.__proxies, verify=self.__verify)
 
             logStr = f"{response.status_code}: {response.reason} {prepStr}"
             APISession.log().debug(f"Response #{tryCount}: {logStr} : {response.text}")

api_utils/auth_factories.py

@@ -0,0 +1,129 @@
+from requests.auth import AuthBase
+from requests import request
+from typing import Dict
+from jsonpath_ng import parse
+import jwt, time, asyncio, logging, json, re
+from datetime import datetime
+from threading import Lock
+from _agent import __agent__
+from api_utils.bearer import HTTPBearerAuth
+from cxone_api.util import json_on_ok
+from cxoneflow_logging import SecretRegistry
+from dataclasses import dataclass, field
+from dataclasses_json import dataclass_json
+
+class AuthFactoryException(BaseException):
+    pass
+
+@dataclass_json
+@dataclass
+class EventContext:
+    raw_event_payload : bytes = field(repr=False)
+    headers : Dict
+    message : Dict = field(init=False)
+
+    def __post_init__(self):
+        self.message = json.loads(self.raw_event_payload)
+
+
+class HeaderFilteredEventContext(EventContext):
+    def __init__(self, raw_event_payload : str, headers : Dict, header_key_regex : str):
+        pattern = re.compile(header_key_regex)
+        EventContext.__init__(self, raw_event_payload=raw_event_payload, headers={k:headers[k] for k in headers if pattern.match(k)})
+
+
+
+
+class AuthFactory:
+    @classmethod
+    def log(clazz):
+        return logging.getLogger(clazz.__name__)
+
+    async def get_auth(self, event_context : EventContext=None, force_reauth : bool=False) -> AuthBase:
+        raise NotImplementedError("get_auth")
+
+    async def get_token(self, event_context : EventContext=None, force_reauth : bool=False) -> str:
+        raise NotImplementedError("get_token")
+
+
+class StaticAuthFactory(AuthFactory):
+    def __init__(self, static_auth : AuthBase):
+        self.__auth = static_auth
+
+    async def get_auth(self, event_context : EventContext=None, force_reauth : bool=False) -> AuthBase:
+        return self.__auth
+
+
+class GithubAppAuthFactory(AuthFactory):
+    __lock = Lock()
+
+    __token_cache = {}
+
+    __event_installation_id = parse("$.installation.id")
+    __app_id_header = "X-Github-Hook-Installation-Target-Id"
+
+
+    def __init__(self, private_key : str, api_url : str):
+        self.__pkey = private_key
+        self.__api_url = api_url
+
+    def __encoded_jwt_factory(self, install_id : int) -> str:
+        payload = {
+            'iat' : int(time.time()),
+            "exp" : int(time.time()) + 600,
+            'iss' : install_id,
+            'alg' : "RS256"
+        }
+
+        return jwt.encode(payload, self.__pkey, algorithm='RS256')
+
+    async def __get_token_tuple(self, event_context : EventContext=None, force_reauth : bool=False):
+
+        install_id_found = GithubAppAuthFactory.__event_installation_id.find(event_context.message)
+        if len(install_id_found) == 0:
+            raise AuthFactoryException("GitHub installation id was not found in the event payload.")
+        install_id = install_id_found[0].value
+
+        if GithubAppAuthFactory.__app_id_header in event_context.headers.keys():
+            app_id = event_context.headers[GithubAppAuthFactory.__app_id_header]
+        else:
+            raise AuthFactoryException(f"Header {GithubAppAuthFactory.__app_id_header} not found in event context.")
+
+        token_tuple = None
+
+        with GithubAppAuthFactory.__lock:
+            if install_id in GithubAppAuthFactory.__token_cache.keys():
+                token_tuple = tuple(GithubAppAuthFactory.__token_cache[install_id])
+                exp = token_tuple[1]
+                if datetime.now(exp.tzinfo) >= exp:
+                    GithubAppAuthFactory.log().debug(f"Token for app_id {app_id} install_id {install_id} expired at {exp}")
+                    token_tuple = None
+
+        if token_tuple is None or force_reauth:    
+            GithubAppAuthFactory.log().debug(f"Generating app token for app_id {app_id} install_id {install_id}")
+            token_response = json_on_ok(await asyncio.to_thread(request, method="POST", 
+                                        url=f"{self.__api_url.rstrip("/")}/app/installations/{install_id}/access_tokens",
+                                        headers = {"User-Agent" : __agent__}, 
+                                        auth=HTTPBearerAuth(self.__encoded_jwt_factory(app_id))))
+            GithubAppAuthFactory.log().debug(f"App token for app_id {app_id} install_id {install_id} generated.")
+
+            token_tuple = (SecretRegistry.register(token_response['token']), datetime.fromisoformat(token_response['expires_at']))
+
+            with GithubAppAuthFactory.__lock:
+                GithubAppAuthFactory.__token_cache[install_id] = token_tuple
+
+        return token_tuple
+
+    async def get_auth(self, event_context : EventContext=None, force_reauth : bool=False) -> AuthBase:
+        if event_context is None:
+            raise AuthFactoryException("Event context is required.")
+        return HTTPBearerAuth ((await self.__get_token_tuple(event_context, force_reauth))[0])
+
+    async def get_token(self, event_context : EventContext=None, force_reauth : bool=False) -> str:
+        if event_context is None:
+            raise AuthFactoryException("Event context is required.")
+        return (await self.__get_token_tuple(event_context, force_reauth))[0]
+
+
+
+

api_utils/bearer.py

@@ -0,0 +1,10 @@
+from requests.auth import AuthBase
+
+class HTTPBearerAuth(AuthBase):
+    def __init__(self, token):
+        AuthBase.__init__(self)
+        self.__token = token
+
+    def __call__(self, r):
+        r.headers["Authorization"] = f"Bearer {self.__token}"
+        return r

api_utils/pagers.py

@@ -0,0 +1,41 @@
+from typing import Callable, Awaitable, List, Any, Dict
+from requests import Response
+
+
+async def async_api_page_generator(coro : Awaitable[Response], 
+                                   data_extractor : Callable[[Response], List], kwargs_gen : Callable[[int], Dict]) -> Any:
+
+    """_summary_
+
+    A generator for paging API calls.
+
+    Args:
+        coro - an awaitable coroutine that will be called with kwargs provideded by kwargs_gen
+
+        data_extractor - A callable that is given a response object returned by coro and is expected to
+        returns a tuple containing elements:
+            0: A list of elements that are provided in the generator.  None or an empty stops the generator.
+            1: A boolean indicating this is the last page.
+
+        kwargs_gen - A method that returns a list used as kwargs when executing coro.  A single int
+        parameter is passed to indicate the current offset count.
+
+    Yields:
+        Any: An extracted object as returned by data_extractor callable.
+    """
+    offset = 0
+    buf = []
+    last_page = False
+
+    while True:
+        if len(buf) == 0 and not last_page:
+            buf, last_page = data_extractor(await coro(**(kwargs_gen(offset))))
+
+            if buf is None or len(buf) == 0:
+                return
+            offset = offset + 1
+        elif len(buf) == 0 and last_page:
+            return
+
+        yield buf.pop()
+