Microsoft Graph API

.github/workflows/publish.yml → .github/workflows/publish-pypi.yml

@@ -1,4 +1,5 @@
-name: publish
+name: Publish to PyPI
+
 on:
   release:
     types:

.github/workflows/publish-test-pypi.yml

@@ -0,0 +1,42 @@
+name: Publish to TestPyPI
+
+on:
+  push:
+    branches:
+      - main
+
+env:
+  PYTHON_VERSION: "3.10"
+
+jobs:
+  publish-test-pypi:
+    name: Publish to TestPyPI
+    runs-on: ubuntu-latest
+    environment:
+      name: test
+      url: https://test.pypi.org/project/kicksaw-integration-utils
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+      - name: Install python
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ env.PYTHON_VERSION }}
+      - name: Cache poetry installation
+        id: cache-poetry
+        uses: actions/cache@v3
+        with:
+          key: poetry-${{ hashFiles('poetry.lock') }}-py${{ env.PYTHON_VERSION }}
+          path: |
+            ~/.local/bin/poetry
+            ~/.local/share/pypoetry
+            ~/.cache/pypoetry
+      - name: Install poetry
+        if: ${{ steps.cache-poetry.outputs.cache-hit != 'true' }}
+        run: curl -sSL https://install.python-poetry.org | python3 -
+      - name: Install project and its dependencies
+        run: poetry install
+      - name: Publish to TestPyPI
+        run: |
+          poetry config repositories.test-pypi https://test.pypi.org/legacy/
+          poetry publish --build --repository test-pypi --username __token__ --password ${{ secrets.PYPI_TOKEN }} --skip-existing

.github/workflows/test.yml

@@ -1,13 +1,11 @@
-name: tests
+name: Test
+
 on:
   pull_request:
   push:
     branches:
       - main
 
-env:
-  PYTHON_VERSION: "3.10"
-
 jobs:
   test:
     name: Test with python-${{ matrix.python-version }}
@@ -41,44 +39,3 @@ jobs:
           poetry install
       - name: Run tests
         run: poetry run pytest --cov-report=xml:coverage.xml
-      - name: Slack Notifications
-        uses: Kicksaw-Consulting/notify-slack-action@master
-        if: always()
-        with:
-          status: ${{ job.status }}
-          notify_when: "failure"
-        env:
-          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
-  deploy-test-pypi:
-    name: Deploy to TestPyPI
-    if: ${{ github.event_name == 'pull_request' && github.event.pull_request.base.ref == 'main' }}
-    needs: test
-    runs-on: ubuntu-latest
-    environment:
-      name: test
-      url: https://test.pypi.org/project/kicksaw-integration-utils
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-      - name: Install python
-        uses: actions/setup-python@v4
-        with:
-          python-version: ${{ env.PYTHON_VERSION }}
-      - name: Cache poetry installation
-        id: cache-poetry
-        uses: actions/cache@v3
-        with:
-          key: poetry-${{ hashFiles('poetry.lock') }}-py${{ env.PYTHON_VERSION }}
-          path: |
-            ~/.local/bin/poetry
-            ~/.local/share/pypoetry
-            ~/.cache/pypoetry
-      - name: Install poetry
-        if: ${{ steps.cache-poetry.outputs.cache-hit != 'true' }}
-        run: curl -sSL https://install.python-poetry.org | python3 -
-      - name: Install project and its dependencies
-        run: poetry install
-      - name: Publish to TestPyPI
-        run: |
-          poetry config repositories.test-pypi https://test.pypi.org/legacy/
-          poetry publish --build --repository test-pypi --username __token__ --password ${{ secrets.PYPI_TOKEN }} --skip-existing

kicksaw_integration_utils/microsoft/graph_api/__init__.py

@@ -0,0 +1,6 @@
+__all__ = [
+    "ApplicationAuth",
+    "DelegatedAuth",
+]
+
+from .auth import ApplicationAuth, DelegatedAuth

kicksaw_integration_utils/microsoft/graph_api/auth.py

@@ -0,0 +1,104 @@
+import datetime
+import json
+
+from abc import ABC, abstractmethod
+from typing import Literal, Optional
+
+import requests
+
+from pydantic import BaseModel, Field, SecretStr
+
+
+# Properties shared by both application and delegated token response
+class AuthTokenBase(BaseModel):
+    token_type: Literal["Bearer"]
+    expires_in: int
+    access_token: SecretStr
+
+
+class ApplicationAuthToken(AuthTokenBase):
+    ext_expires_in: int
+
+    # Helper attributes
+    creation_time: datetime.datetime = Field(default_factory=datetime.datetime.utcnow)
+
+    @property
+    def expiration_time(self) -> datetime.datetime:
+        return self.creation_time + datetime.timedelta(seconds=self.expires_in)
+
+    @property
+    def is_expired(self) -> bool:
+        return (self.expiration_time - datetime.datetime.utcnow()).total_seconds() < 60
+
+
+class DelegatedAuthToken(AuthTokenBase):
+    scope: str
+    refresh_token: SecretStr
+
+
+class AuthBase(ABC):
+    def __init__(self, client_id: str, client_secret: str, tenant_id) -> None:
+        self.tenant_id = tenant_id
+        self.client_id = client_id
+        self.client_secret = client_secret
+
+    @property
+    @abstractmethod
+    def access_token(self) -> AuthTokenBase:
+        pass
+
+    @abstractmethod
+    def refresh_access_token(self) -> None:
+        pass
+
+
+class ApplicationAuth(AuthBase):
+    """Used to access Graph API without a user (server application)."""
+
+    _token: Optional[ApplicationAuthToken]
+
+    @property
+    def access_token(self) -> ApplicationAuthToken:
+        try:
+            token = self._token
+        except AttributeError:
+            self.refresh_access_token()
+            token = self._token
+        else:
+            if token is None or token.is_expired:
+                self.refresh_access_token()
+        assert token is not None
+        return token
+
+    def refresh_access_token(self) -> None:
+        """
+        https://learn.microsoft.com/en-us/graph/auth-v2-service?view=graph-rest-1.0#4-get-an-access-token
+
+        """
+        response = requests.post(
+            url=(
+                f"https://login.microsoftonline.com/{self.tenant_id}"
+                f"/oauth2/v2.0/token"
+            ),
+            data={
+                "client_id": self.client_id,
+                "scope": "https://graph.microsoft.com/.default",
+                "client_secret": self.client_secret,
+                "grant_type": "client_credentials",
+            },
+            timeout=30,
+            headers={
+                "Host": "login.microsoftonline.com",
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+        )
+        response.raise_for_status()
+        self._token = ApplicationAuthToken.parse_obj(json.loads(response.content))
+
+
+class DelegatedAuth(AuthBase):
+    """
+    Used to access Graph API on behalf of a user using OAuth 2.0
+    authorization code grant flow.
+
+    """

pyproject.toml

@@ -21,19 +21,21 @@ boto3 = "^1.17.3"
 pydantic = "^1.9.0"
 
 [tool.poetry.group.test.dependencies]
-pytest = "^7.1.3"
-simple-mockforce = "^0.4.2"
-moto = "^4.0.2"
-pytest-cov = "^3.0.0"
+pytest = "*"
+simple-mockforce = "*"
+moto = "*"
+pytest-cov = "*"
+responses = "*"
 
 [tool.poetry.group.lint.dependencies]
-black = "^22.8.0"
-flake8 = "^5.0.4"
-pylint = "^2.15.0"
-mypy = "^0.971"
+black = "*"
+flake8 = "*"
+pylint = "*"
+mypy = "*"
 
 [tool.poetry.group.typing.dependencies]
 boto3-stubs = {extras = ["essential"], version = "^1.24.66"}
+types-requests = "^2.28.11.2"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]