✨ airbyte-cdk - Adds JwtAuthenticator to low-code

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.82.0
+low-code: Add JWTAuthenticator
+
 ## 0.81.3
 Republish print buffer after previous pypi attempt timed out
 
@@ -31,7 +34,7 @@ Fix CDK version mismatch introduced in 0.78.8
 Update error messaging/type for missing streams. Note: version mismatch, please use 0.78.9 instead
 
 ## 0.78.6
-low-code: add backward compatibility for old close slice behavior 
+low-code: add backward compatibility for old close slice behavior
 
 ## 0.78.5
 low-code: fix stop_condition instantiation in the cursor pagination
@@ -43,7 +46,7 @@ low-code: Add last_record and last_page_size interpolation variables to paginati
 Fix dependencies for file-based extras
 
 ## 0.78.2
-low-code: fix retrieving partition key for legacy state migration 
+low-code: fix retrieving partition key for legacy state migration
 
 ## 0.78.1
 connector-builder: return full url-encoded URL instead of separating parameters
@@ -351,7 +354,7 @@ File CDK: Avoid listing all files for check command
 Vector DB CDK: Expose stream identifier logic, add field remapping to processing | File CDK: Emit analytics message for used streams
 
 ## 0.51.40
-Add filters for base64 encode and decode in Jinja Interpolation 
+Add filters for base64 encode and decode in Jinja Interpolation
 
 ## 0.51.39
 Few bug fixes for concurrent cdk
@@ -680,8 +683,8 @@ Publishing Docker image for source-declarative-manifest
 ## 0.29.0
 **Breaking changes: We have promoted the low-code CDK to Beta. This release contains a number of breaking changes intended to improve the overall usability of the language by reorganizing certain concepts, renaming, reducing some field duplication, and removal of fields that are seldom used.**
 
-The changes are: 
-* Deprecated the concept of Stream Slicers in favor of two individual concepts: Incremental Syncs, and Partition Routers: 
+The changes are:
+* Deprecated the concept of Stream Slicers in favor of two individual concepts: Incremental Syncs, and Partition Routers:
   * Stream will define an `incremental_sync` field which is responsible for defining how the connector should support incremental syncs using a cursor field. `DatetimeStreamSlicer` has been renamed to `DatetimeBasedCursor` and can be used for this field.
   * `Retriever`s will now define a `partition_router` field. The remaining slicers are now called `SubstreamPartitionRouter` and `ListPartitionRouter`, both of which can be used here as they already have been.
   * The `CartesianProductStreamSlicer` because `partition_router` can accept a list of values and will generate that same cartesian product by default.
@@ -860,7 +863,7 @@ Low-code: Fix a few bugs with the stream slicers
 Low-code: Add support for custom error messages on error response filters
 
 ## 0.3.0
-Publish python typehints via `py.typed` file. 
+Publish python typehints via `py.typed` file.
 
 ## 0.2.3
 - Propagate options to InterpolatedRequestInputProvider

airbyte-cdk/python/airbyte_cdk/sources/declarative/auth/__init__.py

@@ -3,7 +3,9 @@
 #
 
 from airbyte_cdk.sources.declarative.auth.oauth import DeclarativeOauth2Authenticator
+from airbyte_cdk.sources.declarative.auth.jwt import JwtAuthenticator
 
 __all__ = [
     "DeclarativeOauth2Authenticator",
+    "JwtAuthenticator"
 ]

airbyte-cdk/python/airbyte_cdk/sources/declarative/auth/jwt.py

@@ -0,0 +1,102 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+from dataclasses import InitVar, dataclass
+from datetime import datetime
+from typing import Any, Mapping, Optional, Union
+
+import jwt
+from airbyte_cdk.sources.declarative.auth.declarative_authenticator import DeclarativeAuthenticator
+from airbyte_cdk.sources.declarative.interpolation.interpolated_mapping import InterpolatedMapping
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+
+
+@dataclass
+class JwtAuthenticator(DeclarativeAuthenticator):
+
+    config: Mapping[str, Any]
+    parameters: InitVar[Mapping[str, Any]]
+    secret_key: Union[InterpolatedString, str]
+    algorithm: Union[InterpolatedString, str]
+    token_duration: Union[InterpolatedString, str] = 1200
+    kid: Union[InterpolatedString, str] = None
+    typ: Union[InterpolatedString, str] = "JWT"
+    iss: Union[InterpolatedString, str] = None
+    sub: Union[InterpolatedString, str] = None
+    aud: Union[InterpolatedString, str] = None
+    cty: Union[InterpolatedString, str] = None
+    additional_jwt_headers: Mapping[str, Any] = None
+    additional_jwt_payload: Mapping[str, Any] = None
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        super().__init__()
+        self._algorithm = InterpolatedString.create(self.algorithm, parameters=parameters)
+        self._secret_key = InterpolatedString.create(self.secret_key, parameters=parameters)
+        self._kid = InterpolatedString.create(self.kid, parameters=parameters)
+        self._typ = InterpolatedString.create(self.typ, parameters=parameters)
+        self._iss = InterpolatedString.create(self.iss, parameters=parameters)
+        self._sub = InterpolatedString.create(self.sub, parameters=parameters)
+        self._aud = InterpolatedString.create(self.aud, parameters=parameters)
+        self._cty = InterpolatedString.create(self.cty, parameters=parameters)
+        self._token_duration = self.token_duration
+        self._additional_jwt_headers = InterpolatedMapping(self.additional_jwt_headers or {}, parameters=parameters)
+        self._additional_jwt_payload = InterpolatedMapping(self.additional_jwt_payload or {}, parameters=parameters)
+
+    def _get_jwt_headers(self) -> Mapping[str, Any]:
+        headers = {}
+        headers.update(self._additional_jwt_headers.eval(self.config))
+        if self._kid:
+            headers["kid"] = f"{self._kid.eval(self.config)}"
+        if self._algorithm:
+            headers["alg"] = self._get_algorithm()
+        if self._typ:
+            headers["typ"] = f"{self._typ.eval(self.config)}"
+        if self._cty:
+            headers["cty"] = f"{self._cty.eval(self.config)}"
+        return headers
+
+    def _get_jwt_payload(self) -> Mapping[str, Any]:
+        payload = {}
+        now = int(datetime.now().timestamp())
+        exp = now + self._token_duration
+        nbf = now
+        payload.update(self._additional_jwt_payload.eval(self.config))
+        if self._iss:
+            payload["iss"] = f"{self._iss.eval(self.config)}"
+        if self._sub:
+            payload["sub"] = f"{self._sub.eval(self.config)}"
+        if self._aud:
+            payload["aud"] = f"{self._aud.eval(self.config)}"
+        payload["iat"] = now
+        payload["exp"] = exp
+        payload["nbf"] = nbf
+        return payload
+
+    def _get_algorithm(self) -> str:
+        algorithm: str = self._algorithm.eval(self.config)
+        if not algorithm:
+            raise ValueError("Algorithm is required")
+        return f"{algorithm}"
+
+    def _get_secret_key(self) -> str:
+        secret_key: str = self._secret_key.eval(self.config)
+        if not secret_key:
+            raise ValueError("secret_key is required")
+        return f"{secret_key}"
+
+    def _get_signed_token(self) -> str:
+        return jwt.encode(
+            payload=self._get_jwt_payload(),
+            key=self._get_secret_key(),
+            algorithm=self._get_algorithm(),
+            headers=self._get_jwt_headers(),
+        )
+
+    @property
+    def auth_header(self) -> str:
+        return "Authorization"
+
+    @property
+    def token(self) -> str:
+        return f"Bearer {self._get_signed_token()}"

airbyte-cdk/python/airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -257,13 +257,15 @@ definitions:
             - "$ref": "#/definitions/BearerAuthenticator"
             - "$ref": "#/definitions/CustomAuthenticator"
             - "$ref": "#/definitions/OAuthAuthenticator"
+            - "$ref": "#/definitions/JwtAuthenticator"
             - "$ref": "#/definitions/NoAuth"
             - "$ref": "#/definitions/SessionTokenAuthenticator"
             - "$ref": "#/definitions/LegacySessionTokenAuthenticator"
         examples:
           - authenticators:
               token: "#/definitions/ApiKeyAuthenticator"
               oauth: "#/definitions/OAuthAuthenticator"
+              jwt: "#/definitions/JwtAuthenticator"
       $parameters:
         type: object
         additionalProperties: true
@@ -833,6 +835,95 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  JwtAuthenticator:
+    title: JWT Authenticator
+    description: Authenticator for requests using JWT authentication flow.
+    type: object
+    required:
+      - type
+      - secret_key
+      - algorithm
+      - jwt_headers
+      - jwt_payload
+    properties:
+      type:
+        type: string
+        enum: [JwtAuthenticator]
+      secret_key:
+        type: string
+        description: Secret used to sign the JSON web token.
+        examples:
+          - "{{ config['secret_key'] }}"
+      algorithm:
+        type: string
+        description: Algorithm used to sign the JSON web token.
+        examples:
+          - "ES256"
+      token_duration:
+        type: integer
+        title: Token Duration
+        description: The amount of time in seconds a JWT token can be valid after being issued.
+        examples:
+          - 1200
+      jwt_headers:
+        type: object
+        title: JWT Headers
+        description: JWT headers used when signing JSON web token.
+        additionalProperties: false
+        properties:
+          kid:
+            type: string
+            title: Key Identifier
+            description: Private key ID for user account.
+            examples:
+              - "{{ config['kid'] }}"
+          typ:
+            type: string
+            title: Type
+            description: The media type of the complete JWT.
+            examples:
+              - JWT
+          cty:
+            type: string
+            title: Content Type
+            description: Content type of JWT header.
+            examples:
+              - JWT
+      additional_jwt_headers:
+        type: object
+        title: Additional JWT Headers
+        description: Additional headers to be included with the JWT headers object.
+        additionalProperties: true
+      jwt_payload:
+        type: object
+        title: JWT Payload
+        description: JWT Payload used when signing JSON web token.
+        additionalProperties: false
+        properties:
+          iss:
+            type: string
+            title: Issuer
+            description: The user/principal that issued the JWT. Commonly a value unique to the user.
+            examples:
+              - "{{ config['iss'] }}"
+          sub:
+            type: string
+            title: Subject
+            description: The subject of the JWT. Commonly defined by the API.
+          aud:
+            type: string
+            title: Audience
+            description: The recipient that the JWT is intended for. Commonly defined by the API.
+            examples:
+              - "appstoreconnect-v1"
+      additional_jwt_payload:
+        type: object
+        title: Additional JWT Payload Properties
+        description: Additional properties to be added to the JWT payload.
+        additionalProperties: true
+      $parameters:
+        type: object
+        additionalProperties: true
   OAuthAuthenticator:
     title: OAuth2
     description: Authenticator for requests using OAuth 2.0 authorization flow.
@@ -1311,6 +1402,7 @@ definitions:
           - "$ref": "#/definitions/BearerAuthenticator"
           - "$ref": "#/definitions/CustomAuthenticator"
           - "$ref": "#/definitions/OAuthAuthenticator"
+          - "$ref": "#/definitions/JwtAuthenticator"
           - "$ref": "#/definitions/NoAuth"
           - "$ref": "#/definitions/SessionTokenAuthenticator"
           - "$ref": "#/definitions/LegacySessionTokenAuthenticator"