Kiteworks API Authentication Flows

GET https://{kiteworks_server}/oauth/authorize
  ?client_id=YOUR_CLIENT_ID
  &response_type=code
  &scope=YOUR_SCOPE
  &redirect_uri=https%3A%2F%2F{kiteworks_server}%2Foauth_callback.php
  HTTP/1.1

HTTP/1.1 302 Found
Location: https://{kiteworks_server}/oauth_callback.php?code=60cc146c8dced75e26e

POST /oauth/token HTTP/1.1
Host: {kiteworks_server}
Content-Type: application/x-www-form-urlencoded

client_id=YOUR_CLIENT_ID
&client_secret=YOUR_CLIENT_SECRET
&grant_type=authorization_code
&code=60cc146c8dced75e26e
&redirect_uri=https%3A%2F%2F{kiteworks_server}%2Foauth_callback.php

{
  "access_token": "d932e1d32d89140163345d47fa97bfa60eeba1a5",
  "expires_in": "360000",
  "token_type": "bearer",
  "scope": "GET\/users\/* *\/files\/*",
  "refresh_token": "d7ce54d721e8das60943f3fc7cb159e4b11d0ee5"
}

#!/usr/bin/env python
# File: python/get_access_token.py

import time
import requests
import webview
from urllib.parse import urlparse, parse_qs
import base64
import hashlib
import os

class KWOAuthClient:
    def __init__(self, base_url):
        self.authorization_endpoint = f"{base_url}/oauth/authorize"
        self.token_endpoint = f"{base_url}/oauth/token"
        self.redirect_uri = f"{base_url}/rest/callback.html"
        self._credential_manager = _CredentialManager()

    @staticmethod
    def _generate_pkce_pair():
        code_verifier = base64.urlsafe_b64encode(os.urandom(32)).rstrip(b'=').decode('utf-8')
        code_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode('utf-8')).digest()
        ).rstrip(b'=').decode('utf-8')
        return code_verifier, code_challenge

    def _is_token_expired(self):
        token_data = self._credential_manager.retrieve_saved_access_token()
        if token_data is None:
            return True
        return int(time.time()) >= token_data['expiry']

    def _refresh_access_token(self):
        refresh_token = self._credential_manager.retrieve_saved_refresh_token()
        if not refresh_token:
            raise ValueError("No refresh token available.")
        data = {
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": self._credential_manager.get_client_id(),
            "client_secret": self._credential_manager.get_client_secret()
        }
        response = requests.post(self.token_endpoint, data=data)
        response.raise_for_status()
        return response.json()

    def _exchange_code_for_tokens(self, code, code_verifier):
        data = {
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": self.redirect_uri,
            "client_id": self._credential_manager.get_client_id(),
            "client_secret": self._credential_manager.get_client_secret(),
            "code_verifier": code_verifier
        }
        response = requests.post(self.token_endpoint, data=data)
        response.raise_for_status()
        return response.json()

    def _save_tokens(self, tokens):
        expires_at = int(time.time()) + tokens.get('expires_in', 3600)
        self._credential_manager.save_tokens(
            tokens['access_token'],
            tokens.get('refresh_token'),
            expires_at
        )

    def _poll_url(self, window, timeout=60):
        start = time.time()
        while True:
            current_url = window.get_current_url()
            if current_url.startswith(self.redirect_uri):
                parsed = urlparse(current_url)
                qs = parse_qs(parsed.query)
                if "code" in qs:
                    self.auth_code = qs["code"][0]
                    window.destroy()
                    break
            if (time.time() - start) > timeout:
                print("Timeout waiting for redirect.")
                window.destroy()
                break
            time.sleep(0.5)

    def _start_oauth_flow(self, code_challenge):
        auth_url = (
            f"{self.authorization_endpoint}"
            f"?client_id={self._credential_manager.client_id}"
            f"&redirect_uri={self.redirect_uri}"
            f"&response_type=code"
            f"&code_challenge={code_challenge}"
            f"&code_challenge_method=S256"
        )
        window = webview.create_window("Login", auth_url)
        webview.start(lambda: self._poll_url(window))
        return self.auth_code

    def get_access_token(self):
        # 1) Return cached token if still valid
        if not self._is_token_expired():
            saved_token = self._credential_manager.retrieve_saved_access_token()
            return saved_token['access_token']

        # 2) Refresh if a refresh token is available
        if self._credential_manager.retrieve_saved_refresh_token() is not None:
            try:
                print("Refreshing access token...")
                tokens = self._refresh_access_token()
                self._save_tokens(tokens)
                return tokens["access_token"]
            except Exception as e:
                print("Failed to refresh token:", e)

        # 3) Full OAuth flow with PKCE
        code_verifier, code_challenge = self._generate_pkce_pair()
        print("Reauthentication required. Opening browser...")
        code = self._start_oauth_flow(code_challenge)
        if not code:
            raise RuntimeError("No authorization code obtained (timeout?).")
        tokens = self._exchange_code_for_tokens(code, code_verifier)
        self._save_tokens(tokens)
        return tokens["access_token"]


class _CredentialManager:
    """
    In-memory credential store for demonstration purposes.
    In production, use a secure secrets manager.
    """
    def __init__(self):
        self.token_data = None
        self.client_id = input("Enter your Client ID: ")
        self.client_secret = input("Enter your Client Secret: ")

    def get_client_id(self):
        return self.client_id

    def get_client_secret(self):
        return self.client_secret

    def save_tokens(self, access_token, refresh_token, expiry):
        self.token_data = {
            'access_token': access_token,
            'refresh_token': refresh_token,
            'expiry': expiry
        }

    def retrieve_saved_access_token(self):
        return self.token_data

    def retrieve_saved_refresh_token(self):
        if self.token_data and 'refresh_token' in self.token_data:
            return self.token_data['refresh_token']
        return None


if __name__ == "__main__":
    base_url = input("Enter Base URL for Kiteworks Instance: ")
    client = KWOAuthClient(base_url)
    token = client.get_access_token()
    print("Got access token:", token)

# Generate a 2048-bit RSA private key
openssl genrsa -out private_key.pem 2048

# Extract the public key
openssl rsa -in private_key.pem -pubout -out public_key.pem

POST /oauth/token HTTP/1.1
Host: kiteworks_server
Content-Type: application/x-www-form-urlencoded

client_id=YOUR_CLIENT_ID
&client_secret=YOUR_CLIENT_SECRET
&grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Ajwt-bearer
&assertion=eyJhbGci...

{
  "access_token": "054915e674bc35fa7fff1f499044e964d3a5d61b",
  "expires_in": 3600,
  "token_type": "bearer",
  "scope": "*/files/* */folders/*"
}

import time
import uuid
import jwt
import requests
from cryptography.hazmat.primitives import serialization

class OAuthJWTAssertionClient:
    """
    OAuth 2.0 JWT Assertion Flow with RSA signatures.

    The Authorization Server expects:
      - client_id
      - client_secret
      - grant_type = "urn:ietf:params:oauth:grant-type:jwt-bearer"
      - assertion  = <signed JWT>

    JWT claims: iss, sub, aud, iat, nbf, exp, jti
    """

    def __init__(self, issuer, subject, audience, private_key,
                 token_endpoint, client_id, client_secret, algorithm="RS256"):
        """
        :param issuer:         The issuer of the JWT (e.g. the Kiteworks domain URL).
        :param subject:        The Kiteworks user account the client acts as.
        :param audience:       The JWT audience — must match the application configuration.
        :param private_key:    RSA private key (PEM) used to sign the JWT.
        :param token_endpoint: URL of the Kiteworks OAuth token endpoint.
        :param client_id:      Registered client application ID.
        :param client_secret:  Registered client secret.
        :param algorithm:      JWT signing algorithm (default: RS256).
        """
        self.issuer = issuer
        self.subject = subject
        self.audience = audience
        self.private_key = private_key
        self.token_endpoint = token_endpoint
        self.client_id = client_id
        self.client_secret = client_secret
        self.algorithm = algorithm

    def create_jwt_assertion(self, validity_seconds=300):
        """
        Create and sign a JWT assertion.
        :param validity_seconds: JWT lifetime in seconds (default: 300 / 5 minutes).
        """
        now = int(time.time())
        payload = {
            "iss": self.issuer,
            "sub": self.subject,
            "aud": self.audience,
            "iat": now,
            "nbf": now,
            "exp": now + validity_seconds,
            "jti": str(uuid.uuid4())
        }
        return jwt.encode(payload, self.private_key, algorithm=self.algorithm)

    def get_access_token(self, scope=None, validity_seconds=300):
        """
        Request an access token using the JWT Bearer assertion.
        :param scope:            Optional scope string.
        :param validity_seconds: JWT lifetime in seconds.
        :return: Access token string.
        """
        assertion = self.create_jwt_assertion(validity_seconds=validity_seconds)
        data = {
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "assertion": assertion,
        }
        if scope:
            data["scope"] = scope

        response = requests.post(self.token_endpoint, data=data)
        response.raise_for_status()

        token_response = response.json()
        access_token = token_response.get("access_token")
        if not access_token:
            raise ValueError("No access_token in response.")
        return access_token