Using CLI/Pystardog with MS Entra ID

Created by Steve Place, Modified on Wed, Apr 16 at 7:27 PM by Steve Place

In this article we will show different ways to use MS Entra ID with Stardog.

Prerequisites

The examples in this article expect that the MSAL library for Python is installed. A similar approach can be used for other languages, as the MSAL library is available for several languages.

We will illustrate how you can print the token or integrate with either pystardog or the Stardog CLI.

Getting a token

Getting a user token (PublicClientApplication)

Often you want to use a CLI against MS Entra ID. Luckily, this is made relatively easy with the MSAL library using two different flows:

Interactive
Device login

The interactive flow requires a browser is installed on the machine where you are running the CLI, as it will redirect you to your browser, where you can authenticate against MS Entra ID.

The device login flow is useful when no browser is installed on the machine where you are running the CLI. When you start the CLI, it will provide a url and a code. Something like:

Go to https://microsoft.com/devicelogin and enter code: AAY4LXAV5

This allows the user to go to a second device at the provided url and enter the code. Once the code is provided, the CLI will receive the requested token.

Before proceeding with the Python code example below, verify in the Azure portal that the Application allows Public flow (under the Authentication tab):

Now you can create a script, example.py, with the following content:

#!/usr/bin/env python3  # you may need to adjust this path


import os
import sys


import msal
import atexit
from pathlib import Path


# Configuration
CLIENT_ID = "ENTER YOUR CLIENT ID"
TENANT_ID = "ENTER YOUR TENANT ID"


DEVICE_LOGIN_FLOW = False
AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}"
SCOPE = [f"api://{CLIENT_ID}/.default"]


arguments = sys.argv[1:]


if '--device-login' in arguments:
   arguments.remove('--device-login')
   DEVICE_LOGIN_FLOW = True


# These are helper to use a cache so the user does not need to authenticate everytime
# the cli is executed, similarly to what a browser does.
CACHE_PATH = os.path.join(Path.home(), '.stardog.token_cache')




def load_token_cache():
   cache = msal.SerializableTokenCache()
   if os.path.exists(CACHE_PATH):
       cache.deserialize(open(CACHE_PATH, 'r').read())
   return cache




def save_token_cache(cache):
   if cache.has_state_changed:
       with open(CACHE_PATH, 'w') as f:
           f.write(cache.serialize())




# Initialize cache
token_cache = load_token_cache()
atexit.register(lambda: save_token_cache(token_cache))  # Save on script exit


# Create a PublicClientApplication
app = msal.PublicClientApplication(
   client_id=CLIENT_ID,
   authority=AUTHORITY,
   token_cache=token_cache
)


# Try to acquire token silently first
accounts = app.get_accounts()
if accounts:
   result = app.acquire_token_silent(SCOPE, account=accounts[0])
else:
   result = None


# If no cached token, use device flow
if not result:


   if DEVICE_LOGIN_FLOW:
       flow = app.initiate_device_flow(scopes=SCOPE)
       if "user_code" not in flow:
           raise ValueError("Failed to create device flow")


       print(f"Go to {flow['verification_uri']} and enter code: {flow['user_code']}")
       result = app.acquire_token_by_device_flow(flow)
   else:
       result = app.acquire_token_interactive(scopes=SCOPE)


# Check and use the token
if "access_token" in result:
   # Access token acquired
   token = result["access_token"]


   print(f"Your token: {token}") # Replace with your own logic
else:
   print("Failed to get token:")
   print(result.get("error"))
   print(result.get("error_description"))

The code above supports both flows. By default, it uses the interactive flow; however, if you give the script the --device-login option, it will use the device login flow.

Our example simply prints the token; however, please see the section on how you can use this token with the Stardog’s CLI or pystardog.

Getting a service principal token

In many cases, you need to perform actions without any human intervention. This can be accomplished by creating an application with a service principle.

From the Azure Portal, ensure you can add the application to the roles that need to accomplish this. This can be done from the manifest. It should look something like the following:

  "appRoles": [
    {
      "allowedMemberTypes": [
        "User",
        "Application"
      ],
      "description": "admin",
      "displayName": "admin",
      "id": "97932c93-52c2-4561-a7a7-7fea2443c6d7",
      "isEnabled": true,
      "origin": "Application",
      "value": "admin"
    },
            . . .
  ],

Once that is completed, you can assign the application to the roles. Note this is not possible if your MS Entra ID level plan is Free or Basic, since you are only allowed to assign individual users to a role on those plans.

Now you can create a script with the following content:

#!/usr/bin/env python3  # you may need to adjust this path


import msal
import requests


# Configuration ====
CLIENT_ID = "ENTER YOUR CLIENT ID"
TENANT_ID = "ENTER YOUR TENANT ID"
CLIENT_SECRET = "ENTER YOUR CLIENT SECRET" # this is sensitive info


AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}"
SCOPE = [f"api://{CLIENT_ID}/.default"]


# Create the confidential client app ====
app = msal.ConfidentialClientApplication(
   client_id=CLIENT_ID,
   authority=AUTHORITY,
   client_credential=CLIENT_SECRET
)


# Acquire token for Graph ====
result = app.acquire_token_silent(SCOPE, account=None)


if not result:
   result = app.acquire_token_for_client(scopes=SCOPE)


# Use the token ====
if "access_token" in result:
   token = result["access_token"]


   print(f"Confidential token: {token}")
else:
   print("Failed to acquire token:")
   print(result.get("error"))
   print(result.get("error_description"))

The CLIENT_SECRET is sensitive information, so keep this script well-protected. If your company has a mechanism to retrieve this value from a Key Vault and ensure it's rotated, please do so and adjust the code accordingly.

This example again simply prints the token. Please see below on how to integrate with Stardog’s CLI or pystardog.

Integration

Stardog CLI

You can adjust the code provided above to call stardog or stardog-admin as needed. This can be accomplished since we can pass the token via the --token options. In other words, the code provided simply acts as a wrapper.

For example, replace the line that print the tokens with the following:

arguments.append("--token")
arguments.append(token)
arguments.insert(0, '/usr/local/bin/stardog-admin') #may need to adjust path


subprocess.call(arguments)

This will call stardog-admin with the argument that you pass on the command line. A similar approach can be done with the stardog command as well.

To install the Stardog CLI, see here.

pystardog

You may want to write your own customized CLI using pystardog, which fully supports an oAuth connection. This can be achieved by adding the following code after the Configuration block:

@dataclasses.dataclass(frozen=True)
class BearerAuth(requests.auth.AuthBase):
   token: str


   def __call__(self, r):
       r.headers["Authorization"] = f"bearer {self.token}"
       return r

and replacing the block in if "access_token" in result: with:

with stardog.Admin(
       endpoint="https://sparql.profile-a.sd-testlab.com",
       auth=BearerAuth(result['access_token'])
) as admin:
   print("Databases:", admin.databases())

The above is a simple example to show the connection aspect only. Adjust accordingly to perform the logic you require.

HTTP API

If you wish, you can also use the HTTP API directly by replacing the block in if "access_token" in result: with:

r = requests.get(url=f'https://{endpoint}/admin/databases', headers=header)
print(r.content)

The above is a simple example to show the connection aspect only. Adjust accordingly to perform the logic you require.