Share code or advice about how you've used the STIGMan API! #497
Replies: 3 comments 8 replies
-
|
PowerShell using stigman-watcher client_id - This script should get the asset information from STIG-Manager when provided the Collection and the asset FQDN. |
Beta Was this translation helpful? Give feedback.
-
|
Anyone interested in how to install Stig Manger in Azure? We have it successfully running with CAC authentication. Stig manager is running in an Azure container instance, HTTPS is handled using an Azure Application Gateway, OIDC is Azure AD (via app registration). There is little documentation online on how to set this all up in Azure. |
Beta Was this translation helpful? Give feedback.
-
|
Sure I'll join in |
Beta Was this translation helpful? Give feedback.
-
|
@mcklutty35 we are currently trying to setup stig manager in our environment using Azure AD for OIDC and running into some issues. Do you have any resources you would be willing to share to nudge us along? |
Beta Was this translation helpful? Give feedback.
-
|
What sort of issues are you having? This is how I set it up in our Azure subscription:
|
Beta Was this translation helpful? Give feedback.
-
|
@mcklutty35 thanks for that. I was missing STIGMAN_JWT_PRIVILEGES_CLAIM = roles. Works now |
Beta Was this translation helpful? Give feedback.
-
|
@mcklutty35 Would you be able to share more details about how you have that setup? I have been trying to get it to work using azure ad but having it run locally in docker desktop for testing. If you have any setup docs I can be reached on flankspeed: [email protected]. |
Beta Was this translation helpful? Give feedback.
-
|
Any samples out there for accessing the APIs in python? Just need a sample for one of the APIs and I can build from there |
Beta Was this translation helpful? Give feedback.
-
|
Getting a JWT access token from the OIDC provider is the first step because you need to send that token with every API request. There are several flows your provider may implement and some they may not support. I'm not a python dev, so I can't evaluate the OIDC libraries you might consider using. But here's a very basic one file implementation I hacked out that uses OIDC Device Authorization flow to get the access token. It uses URLs that work with our Quick Start Orchestration on Docker Hub. Most of the code is authentication, calling the API is pretty straightforward after that. #!/usr/bin/python3
import concurrent.futures
import logging
import json
import requests
import time
import datetime
import webbrowser
from urllib.parse import urljoin
log = logging.getLogger(__name__)
def main():
print("STIG Manager API with python example")
session = Session(
client_id='stig-manager',
oidc_provider_base='http://localhost:8080/realms/stigman/',
api_base='http://localhost:54000/api/'
)
session.login_device_auth()
print('Getting this user\'s information:')
response = session.request('GET', 'user')
print(json.dumps(response.json(), indent=2))
print('Getting collections visible to this user:')
response = session.request('GET', 'collections', params='projection=labels&projection=statistics')
print(json.dumps(response.json(), indent=2))
class LinkLogin(object):
"""
The data required for prompting the user during OIDC Device Authorization flow
"""
#: Amount of seconds until the code expires
expires_in = None
#: The code the user should enter at the uri
user_code = None
#: The link the user has to visit
verification_uri = None
#: The link the user has to visit, with the code already included
verification_uri_complete = None
def __init__(self, json):
self.expires_in = json['expires_in']
self.user_code = json['user_code']
self.verification_uri = json['verification_uri']
self.verification_uri_complete = json['verification_uri_complete']
class Session(object):
#: The STIG Manager API access token, this is what you use with basic_request
access_token = None
#: A :class:`datetime` object containing the date the access token will expire
expiry_time = None
#: A refresh token for retrieving a new access token through refresh_token
refresh_token = None
#: The type of access token, e.g. Bearer
token_type = None
#: The OIDC Provider's base URL, with trailing slash
oidc_provider_base = None
#: The client_id for your client with the OIDC Provider. Use 'stig-manager' to emulate the project's web app
client_id = None
#: The base URL of the STIG Manager API service, with trailing slash
api_base = None
#: The token scopes to request. Remove :read to request write access
scope = 'openid stig-manager:collection:read stig-manager:stig:read stig-manager:user:read stig-manager:op:read'
def __init__(self, client_id=None, oidc_provider_base=None, api_base=None):
self.client_id = client_id
self.oidc_provider_base = oidc_provider_base
self.api_base = api_base
def _get_oidc_metadata(self):
url = urljoin(self.oidc_provider_base, '.well-known/openid-configuration')
response = requests.get(url)
json = response.json()
if not response.ok:
log.error("Fetching OIDC metadata failed: %s", response.text)
response.raise_for_status()
self.device_authorization_endpoint = json['device_authorization_endpoint']
self.token_endpoint = json['token_endpoint']
def login_device_auth(self, function=print):
"""
Login to STIG Manager using OIDC Device Authorization flow. You can select what function you want to use to display the link
:param function: The function you want to display the link with
:raises: TimeoutError: If the login takes too long
"""
self._get_oidc_metadata()
login, future = self._login_with_link()
text = "Visit {0} to log in, the code will expire in {1} seconds"
function(text.format(login.verification_uri_complete, login.expires_in))
webbrowser.open(login.verification_uri_complete)
future.result()
def _login_with_link(self):
params = {
'client_id': self.client_id,
'scope': self.scope
}
response = requests.post(self.device_authorization_endpoint, params)
if not response.ok:
log.error("Login failed: %s", response.text)
response.raise_for_status()
json = response.json()
executor = concurrent.futures.ThreadPoolExecutor()
return LinkLogin(json), executor.submit(self._process_link_login, json)
def _process_link_login(self, json):
json = self._wait_for_link_login(json)
self.access_token = json['access_token']
self.expiry_time = datetime.datetime.now() + datetime.timedelta(seconds=json['expires_in'])
self.refresh_token = json['refresh_token']
self.token_type = json['token_type']
def _wait_for_link_login(self, json):
expiry = json['expires_in']
interval = json['interval']
device_code = json['device_code']
url = self.token_endpoint
params = {
'client_id': self.client_id,
'device_code': device_code,
'grant_type': 'urn:ietf:params:oauth:grant-type:device_code',
'scope': self.scope
}
while expiry > 0:
response = requests.post(url, params)
json = response.json()
if response.ok:
return json
# Because the requests take time, the expiry variable won't be accurate, so stop if OP says it's expired
if json['error'] == 'expired_token':
break
time.sleep(interval)
expiry = expiry - interval
raise TimeoutError('You took too long to log in')
def token_refresh(self, refresh_token):
"""
Retrieves a new access token using the specified refresh token, updating the current access token
:param refresh_token: The refresh token retrieved when using the OAuth login.
:return: True if we believe the token was successfully refreshed, otherwise False
"""
url = self.token_endpoint
params = {
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
'client_id': self.client_id
}
response = requests.post(url, params)
json = response.json()
if not response.ok:
log.warning("The refresh token has expired, a new login is required.")
return False
self.access_token = json['access_token']
self.expiry_time = datetime.datetime.now() + datetime.timedelta(seconds=json['expires_in'])
self.token_type = json['token_type']
return True
def basic_request(self, method, path, params=None, data=None, headers=None):
if not headers:
headers = {}
if self.token_type:
headers['authorization'] = self.token_type + ' ' + self.access_token
url = urljoin(self.api_base, path)
response = requests.request(method, url, params=params, data=data, headers=headers)
if not response.ok and response.json()['message'].startswith("jwt expired") and self.refresh_token:
log.debug("The access token has expired, trying to refresh it.")
refreshed = self.token_refresh(self.refresh_token)
if refreshed:
return self.basic_request(method, path, params, data, headers)
return response
def request(self, method, path, params=None, data=None, headers=None):
response = self.basic_request(method, path, params, data, headers)
log.debug("request: %s", response.request.url)
if not response.ok:
print(response.text)
response.raise_for_status()
if response.content:
log.debug("response: %s", json.dumps(response.json(), indent=4))
return response
if __name__ == '__main__':
main() |
Beta Was this translation helpful? Give feedback.
-
If you've written your own client to make use of our API, or have any tips or tricks for other users, post them below!
If you can post the repo of your code publicly, even better! Share a link below!
Thanks for using STIG Manager OSS.
Beta Was this translation helpful? Give feedback.
All reactions