PTFE-1483 support of authentication with github app (#165)

.github/workflows/main.yaml

@@ -26,6 +26,9 @@ jobs:
           cache: pip
       - name: Install dependencies
         run: pip install -r requirements.txt
+      - name: Boot compose test service
+        run: docker compose up --build --detach
+        working-directory: bert_e/tests/images
       - name: Install tox
         run: pip install tox
       - run: tox -e utests

bert_e/bert_e.py

@@ -42,19 +42,22 @@ def __init__(self, settings):
             settings.repository_host,
             settings.robot.username,
             settings.robot_password,
-            settings.robot_email
+            settings.robot_email,
+            settings.github_app_id,
+            settings.github_installation_id,
+            settings.github_private_key,
         )
-        self.project_repo = self.client.get_repository(
-            owner=settings.repository_owner,
-            slug=settings.repository_slug
-        )
-        settings['use_queue'] = not settings.disable_queues
         if settings.repository_host == 'bitbucket':
             self.settings.robot.account_id = self.client.get_user_id()
         if settings.repository_host == 'github':
             if settings['tasks']:
                 LOG.warning("Disabling tasks on GitHub repo")
                 settings['tasks'] = []
+        self.project_repo = self.client.get_repository(
+            owner=settings.repository_owner,
+            slug=settings.repository_slug
+        )
+        settings['use_queue'] = not settings.disable_queues
 
         self.git_repo = GitRepository(
             self.project_repo.git_url,

bert_e/git_host/bitbucket/__init__.py

@@ -51,7 +51,8 @@ def build_filter_query(filters):
 
 @factory.api_client('bitbucket')
 class Client(base.BertESession, base.AbstractClient):
-    def __init__(self, bitbucket_login, bitbucket_password, bitbucket_mail):
+    def __init__(self, bitbucket_login, bitbucket_password, bitbucket_mail,
+                 *args, **kwargs):
         super().__init__()
         headers = {
             'Accept': 'application/json',
@@ -63,6 +64,8 @@ def __init__(self, bitbucket_login, bitbucket_password, bitbucket_mail):
         self.email = bitbucket_mail
         self.headers.update(headers)
         self.auth = HTTPBasicAuth(bitbucket_login, bitbucket_password)
+        self.args = args
+        self.kwargs = kwargs
 
     def get_repository(self, slug, owner=None):
         """Get the repository with the associated owner and slug."""

bert_e/git_host/github/__init__.py

@@ -13,8 +13,11 @@
 # limitations under the License.
 import json
 import logging
+import time
+from functools import lru_cache
 from collections import defaultdict, namedtuple
 from itertools import groupby
+from jwt import JWT, jwk_from_pem
 
 from requests import HTTPError
 from urllib.parse import quote_plus as quote
@@ -37,27 +40,84 @@ class Error(base.Error):
 @factory.api_client('github')
 class Client(base.AbstractClient):
 
-    def __init__(self, login: str, password: str, email: str, org=None,
-                 base_url='https://api.github.com'):
-        headers = {
-            'Accept': 'application/vnd.github.v3+json',
-            'User-Agent': 'Bert-E',
-            'Content-Type': 'application/json',
-            'From': email,
-            'Authorization': 'token ' + password
-        }
+    def __init__(self, login: str, password: str, email: str,
+                 app_id: int | None = None, installation_id: int | None = None,
+                 private_key: str | None = None, org=None,
+                 base_url='https://api.github.com',
+                 accept_header="application/vnd.github.v3+json"):
 
         rlog = logging.getLogger('requests.packages.urllib3.connectionpool')
         rlog.setLevel(logging.CRITICAL)
         self.session = base.BertESession()
-        self.session.headers.update(headers)
 
         self.login = login
         self.password = password
+        self.app_id = app_id
+        self.installation_id = installation_id
+        self.private_key = jwk_from_pem(private_key)
         self.email = email
         self.org = org
         self.base_url = base_url.rstrip('/')
         self.query_cache = defaultdict(LRUCache)
+        self.accept_header = accept_header
+
+        self.session.headers.update(self.headers)
+
+    def _get_jwt(self):
+        """Get a JWT for the installation."""
+
+        payload = {
+            # Issued at time
+            'iat': int(time.time()),
+            # JWT expiration time (10 minutes maximum)
+            'exp': int(time.time()) + 600,
+            # GitHub App's identifier
+            'iss': self.app_id
+        }
+        jwt_instance = JWT()
+        return jwt_instance.encode(payload, self.private_key, alg='RS256')
+
+    @lru_cache()
+    def _get_installation_token(self, ttl_cache=None):
+        """Get an installation token for the client's installation."""
+        # ttl_cache is a parameter used by lru_cache to set the time to live
+        # of the cache. It is not used in this method.
+        del ttl_cache
+
+        url = (
+            f'{self.base_url}/app/installations/'
+            f'{self.installation_id}/access_tokens'
+        )
+        headers = {
+            'Authorization': f'Bearer {self._get_jwt()}',
+            'Accept': self.accept_header,
+        }
+        print(headers)
+        response = self.session.post(url, headers=headers)
+        response.raise_for_status()
+        return response.json()['token']
+
+    @property
+    def is_app(self):
+        if self.app_id and self.installation_id and self.private_key:
+            return True
+        return False
+
+    @property
+    def headers(self):
+        headers = {
+            'Accept': self.accept_header,
+            'User-Agent': 'Bert-E',
+            'Content-Type': 'application/json',
+            'From': self.email,
+        }
+        if self.is_app:
+            token = self._get_installation_token(
+                ttl_cache=round(time.time() / 600))
+            headers['Authorization'] = f'Bearer {token}'
+        else:
+            headers['Authorization'] = f'token {self.password}'
+        return headers
 
     def _patch_url(self, url):
         """Patch URLs if it is relative to the API root.

bert_e/git_host/mock.py

@@ -55,11 +55,13 @@ class Error404Response:
 
 @api_client('mock')
 class Client(base.AbstractClient):
-    def __init__(self, username, password, email):
+    def __init__(self, username, password, email, *args, **kwargs):
         self.login = username
         self.password = password
         self.auth = self
         self.email = email
+        self.args = args
+        self.kwargs = kwargs
 
     def create_repository(self, slug, owner=None, scm='git',
                           is_private=True):

bert_e/settings.py

@@ -190,6 +190,9 @@ class Meta:
 
     client_id = fields.Str(required=False, load_default='')
     client_secret = fields.Str(required=False, load_default='')
+    github_app_id = fields.Int(required=False, load_default='')
+    github_private_key = fields.Str(required=False, load_default='')
+    github_installation_id = fields.Int(required=False, load_default='')
 
     @pre_load(pass_many=True)
     def load_env(self, data, **kwargs):

bert_e/tests/images/compose.yaml

@@ -0,0 +1,6 @@
+---
+services:
+  github-mock:
+    build: github-mock
+    ports:
+      - 4010:4010

bert_e/tests/images/github-mock/Dockerfile

@@ -0,0 +1,29 @@
+FROM debian:bookworm-slim
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+    ca-certificates \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+ENV PRISM_VERSION=v5.3.1
+
+RUN curl https://github.com/stoplightio/prism/releases/download/${PRISM_VERSION}/prism-cli-linux \
+    -L \
+    -o /usr/local/bin/prism \
+    && chmod +x /usr/local/bin/prism
+
+WORKDIR /app
+
+RUN curl -O -L https://raw.githubusercontent.com/github/rest-api-description/main/descriptions/ghec/ghec.2022-11-28.json
+
+# There's a misconfiguration in the openapi file, we are going to replace the following strings:
+# - "server-statistics-actions.yaml" -> "#/components/schemas/server-statistics-actions"
+# - "server-statistics-packages.yaml" -> "#/components/schemas/server-statistics-packages"
+
+RUN sed -i 's/server-statistics-actions.yaml/#\/components\/schemas\/server-statistics-actions/g' ghec.2022-11-28.json \
+    && sed -i 's/server-statistics-packages.yaml/#\/components\/schemas\/server-statistics-packages/g' ghec.2022-11-28.json
+
+ENTRYPOINT [ "prism" ]
+
+CMD ["mock", "ghec.2022-11-28.json", "-h", "0.0.0.0"]

bert_e/tests/unit/test_github_app_auth.py

@@ -0,0 +1,35 @@
+from bert_e.git_host.github import Client
+from pytest import fixture
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.backends import default_backend
+
+
+@fixture
+def client_app():
+    private_key = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+        backend=default_backend()
+    )
+    return Client(
+        login='login',
+        password='password',
+        email='[email protected]',
+        app_id=1,
+        installation_id=1,
+        private_key=private_key.private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption()
+        ),
+        base_url="http://localhost:4010",
+        accept_header="application/json"
+    )
+
+
+def test_github_auth_app(client_app):
+    repository = client_app.get_repository('octo-org', 'Hello-World')
+    pr = repository.get_pull_request(1)
+    assert pr.id == 1347
+    assert client_app.headers['Authorization'].startswith('Bearer ') is True

requirements.txt

@@ -18,3 +18,4 @@ requests==2.28.2
 requests-mock==1.10.0
 werkzeug==2.2.3
 WTForms==3.0.1
+jwt==1.3.1