Plug apps back into Lumos using an integration connector built with this SDK.
pip install connector-py
This package has...
To get started with the CLI, run connector --help
This SDK has an OpenAPI spec that you can render and view with the Swagger editor.
connector spec
From your shell, run
# Create a connector
# CLI cmd name folder
connector scaffold demo-connector demo_connector
# Install its dependencies in a virtual env
cd demo_connector
python -m venv .venv
. .venv/bin/activate
pip install .
# Lint and run tests
mypy .
pytest
# Run the info capability (note the hyphens, instead of underscores)
demo-connector info
Custom and on-premise Lumos connectors are called via the CLI; they're passed JSON and should print the response JSON to stdout.
Run the info capability to learn what other capabilities the connector supports, what resource and entitlement types, its name, etc.
Look at the info, using jq to pretty-print the response:
demo-connector info | jq .response
# or just the capabilities
demo-connector info | jq .response.capabilities
To call most capabilities, you run a command where you pass the request (JSON) as a string.
<CONNECTOR COMMAND> <CAPABILITY NAME> --json '<A STRINGIFIED JSON OBJECT>'
The most important capability to implement is validate_credentials. Lumos uses this capability to ensure a user-established connection works, and has resulted in authentication credentials your connector can use to perform other actions.
test-connector validate_credentials --json '{
"auth": {
"oauth": {
"access_token":"this will not work"
}
},
"request": {},
"settings": {
"account_id":"foo"
}
}'
This is expected to 💥 fail with a brand-new connector. You'll need to figure out how to configure the authentication to the underlying app server, and how to surface that as user (auth) configuration.
To learn more about all the capabilities, check out the OpenAPI spec in a Swagger editor.
To see a working capability, you can use the Lumos mock connector's validate_credentials call, using jq to pretty print the JSON:
mock-connector validate_credentials --json '{"auth":{"basic":{"username":"foo","password":"bar"}},"request":{},"settings":{"host":"google.com"}}' | jq .
{
"response": {
"valid": true,
"unique_tenant_id": "mock-connector-tenant-id"
},
"page": null
}
Connectors can implement whichever Lumos capabilities make sense for the underlying app.
To see what a minimal implementation looks like, you can inspect a newly scaffolded connector, and look at the integration declaration, and the uncommented out capability registrations.
The integration declaration looks something like this:
integration = Integration(
app_id="my_app",
version=__version__,
auth=BasicCredential,
settings_model=MyAppSettings,
exception_handlers=[
(httpx.HTTPStatusError, HTTPHandler, None),
],
description_data=DescriptionData(
logo_url="https://logo.clearbit.com/foobar.com",
user_friendly_name="Foo Bar",
description="Foobar is a cloud-based platform that lets you manage foos and bars",
categories=[AppCategory.DEVELOPERS, AppCategory.COLLABORATION],
),
resource_types=resource_types,
entitlement_types=entitlement_types,
)
And capability registration looks something like this:
@integration.register_capability(CapabilityName.LIST_ACCOUNTS)
async def list_accounts(request: ListAccountsRequest) -> ListAccountsResponse:
# do whatever is needed to get accounts, probably make an http call
return ListAccountsResponse(
response=[],
...
)
Scaffolded connectors come with a bunch of unit test examples - they're all skipped by default, but you can remove the skip marker to use the existing test.
To run unit tests:
pytest .
To understand the test structure:
demo_connector/
demo_connector/
tests/
test_basic_capabilities/
test_list_accounts_cases.py
test_list_accounts.py
...
test_list_accounts.py is the actual Pytest test code. It uses all the existing test cases fromtest_list_accounts_cases.pyYou can see the reference here
@pytest_cases.parametrize_with_cases(
["args", "response_body_map", "expected_response"],
cases=[
# The name of the Python module
"tests.test_basic_capabilities.test_list_accounts_cases",
],
)
The generated Python code is typed, and can be typechecked with MyPy (installed as a dev dependency).
mypy .
Error handling is facilitated through an exception handler decorator.
An exception handler can be attached to the connector library as follows:
from httpx import HTTPStatusError
from connector.oai.errors import HTTPHandler
integration = Integration(
...,
exception_handlers=[
(HTTPStatusError, HTTPHandler, None),
],
handle_errors=True,
)
The decorator accepts a list of tuples of three.
By default it is recommended to make use of the default HTTPHandler which will handle raise_for_status() for you and properly error code it. For more complex errors it is recommended to subclass the ExceptionHandler (in connector/oai/errors.py) and craft your own handler.
Among this, there is a custom exception class available as well as a default list of error codes:
from connector.oai.errors import ConnectorError
from connector.generated import ErrorCode
def some_method(self, args):
raise ConnectorError(
message="Received wrong data, x: y",
app_error_code="foobar.some_unique_string",
error_code=ErrorCode.BAD_REQUEST,
)
It is preferred to raise any manually raisable exception with this class. A connector can implement its own error codes list, which should be properly documented.
An example response when handled this way:
// BAD_REQUEST error from github connector
{"error":{"message":"Some message","status_code":400,"error_code":"bad_request","raised_by":"HTTPStatusError","raised_in":"github.integration:validate_credentials"}, "response": null, "raw_data": null}
The OAuth module is responsible for handling the OAuth2.0 flow for a connector.
It is configured with oauth_settings in the Integration class.
Not configuring this object will disable the OAuth module completely.
from connector.oai.modules.oauth_module_types import (
OAuthSettings,
OAuthCapabilities,
OAuthRequest,
RequestDataType,
)
integration = Integration(
...,
oauth_settings=OAuthSettings(
# Authorization & Token URLs for the particular connector
authorization_url="https://app.connector.com/oauth/authorize",
token_url="https://api.connector.com/oauth/v1/token",
# Scopes per capability (space delimited string)
scopes={
CapabilityName.VALIDATE_CREDENTIALS: "test:scope another:scope",
... # further capabilities as implemented in the connector
},
# You can modify the request type if the default is not appropriate
# common options for method are "POST" and "GET"
# available options for data are "FORMDATA", "QUERY", and "JSON" (form-data / url query params / json body)
# *default is POST and FORMDATA*
request_type=OAuthRequest(data=RequestDataType.FORMDATA),
# You can modify the authentication method if the default is not appropriate
# available options for auth_method are "CLIENT_SECRET_POST" and "CLIENT_SECRET_BASIC"
# *default is CLIENT_SECRET_POST*
client_auth=ClientAuthenticationMethod.CLIENT_SECRET_POST,
# You can turn off specific or all capabilities for the OAuth module
# This means that these will either be skipped or you have to implement them manually
capabilities=OAuthCapabilities(
refresh_access_token=False,
),
# You can specify the type of OAuth flow to use
# Available options are "CODE_FLOW" and "CLIENT_CREDENTIALS"
# *default is CODE_FLOW*
flow_type=OAuthFlowType.CODE_FLOW,
),
)
It might happen that your integration requires a dynamic authorization/token URL.
For example when the service provider has specific URLs and uses the customers custom subdomain. (eg. https://{subdomain}.service.com/oauth/authorize)
In that case you can pass a callable that takes the request args (AuthRequest, without the auth parameter) as an argument (only available during request).
# method definitions
def get_authorization_url(args: AuthRequest) -> str:
settings = get_settings(args, ConnectorSettings)
return f"https://{settings.subdomain}.service.com/oauth/authorize"
def get_token_url(args: AuthRequest) -> str:
settings = get_settings(args, ConnectorSettings)
return f"https://{settings.subdomain}.service.com/oauth/token"
# oauth settings
integration = Integration(
...,
oauth_settings=OAuthSettings(
authorization_url=get_authorization_url,
token_url=get_token_url,
),
)
The OAuth module supports two flow types:
CODE_FLOW: The authorization code flow (default)CLIENT_CREDENTIALS: The client credentials flow (sometimes called "2-legged OAuth" or "Machine-to-Machine OAuth")The flow type can be specified in the OAuthSettings object.
Using the authorization code flow you have three available capabilities:
GET_AUTHORIZATION_URL: To get the authorization URLHANDLE_AUTHORIZATION_CALLBACK: To handle the authorization callbackREFRESH_ACCESS_TOKEN: To refresh the access tokenUsing the client credentials flow you have two available capabilities:
HANDLE_CLIENT_CREDENTIALS_REQUEST: To handle the client credentials request, uses the token URLREFRESH_ACCESS_TOKEN: To refresh the access tokenThese are registered by default via the module and can be overriden by the connector.
If you run:
connector info
You will see that the OAuth capabilities are included in the available connector capabilities.
A connector is used to connect to multiple tenants of the same app. Each tenant has a connection in Lumos, and the unique tenant ID is used to distinguish the different connections.
Each connection has its own...
A connector can be used for multiple underlying instances of the same app. For instance, you might use a github connector to establish connections with different Github Organizations. The nature of "what is a tenant" is dependent on the underlying app.
A scaffolded connector has OAuth authentication, and a Settings type with account_id. You don't have to keep these - you can change the authentication model and the Settings type to whatever is appropriate for the underlying app (settings may be empty).
A quick rule is sensitive data that would allow an attacker the ability to access the underlying app, goes into the auth payload. Anything else that's not sensitive, not absolutely required to connect to a tenant, or can have a sane default, is settings.
info command.info response to render a connection form for the user.
validate_credentials capability.
At this point, the connection is considered established, and Lumos will attempt to read all data from the connector, allow user provisioning and deprovisioning, etc.
Quick steps:
Package up the connector you've built into a single executable. We use pyinstaller for our Python connectors.
# SDK command module name package dir
connector compile-on-prem demo_connector ./demo_connector
Run the Lumos on-premise agent.
On the same machine as (3), deploy the packaged-up connector from (1) in the same folder.
The integration should show up in the Lumos AdminView > Integrations screen.
Lumos calls a connector's APIs with auth and settings data to read all the accounts, entitlements, resources, and associations in the connected app.
There are two ways this happens, depending on who's hosting the connector.
If Lumos is hosting it, we call it directly in our backend.
If it's a custom connector, it runs as an on-premise connector on a customer's computer, and is called by the Lumos on-prem agent.
You can use a package called asgiref. This package converts I/O bound synchronous
calls into asyncio non-blocking calls. First, add asgiref to your dependencies list
in pyproject.toml. Then, in your async code, use asgiref.sync_to_async to convert
synchronous calls to asynchronous calls.
from asgiref.sync import sync_to_async
import requests
async def async_get_data():
response = await sync_to_async(requests.get)("url")
connector is distributed under the terms of the Apache 2.0 license.
FAQs
An Abstract Tool to Perform Actions on Integrations
We found that connector-py demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.