Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Custom AWS SSM-based device profile persistor implementation for the Ionic Python SDK
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import division, print_function
import json
import os
import boto3
import ionicsdk
# From:
import custom_persistors as custom_persistors
class ParameterStorePersistor(custom_persistors.DeviceProfilePersistorBase):
Store profiles in a subpath of AWS param store.
Tag filters should be a list of objects of the following form
"Key": "string",
"Values": [ "string" ]
Example usage
from cloud_persistors import ParameterStorePersistor
from custom_persistors import Agent
# Load profiles from paramstore
aa = Agent(None, ParameterStorePersistor("/ionic/auth/timothy/devdot", [{"Key": "Type", "Values": ["IonicProfile"]}]), True)
def __init__(self, prefix="/", tag_filters=None):
Prefix is the prefix to search under.
Tags are the set of tags marking SEPs.
self.prefix = prefix
self.tag_filters = tag_filters or []
def loadprofiles(self, rename_imported_profiles=True, *args, **kwargs):
Load profiles from a path on paramstore.
If `rename_imported_profiles` is True (the default), profiles will be renamed based on their source.
ssm_client = boto3.client('ssm')
# Assumes all values are encrypted purposefully so that if the user tries to use this with non-encrypted
# parameters they will receive an error
raw_params = []
if self.tag_filters:
# We would prefer to use 'get_parameters' here, but that API does not support filtering by tag.
parameter_filters = []
if self.tag_filters:
for filter_tag in self.tag_filters:
Key="tag:%s" % (filter_tag["Key"]),
for pp in ssm_client.describe_parameters(ParameterFilters=parameter_filters)['Parameters']:
raw_params.append(ssm_client.get_parameter(Name=pp['Name'], WithDecryption=True)["Parameter"])
# We don't need tag filtering; use the simpler api
raw_params = ssm_client.get_parameters_by_path(Path=self.prefix, WithDecryption=True, Recursive=True)['Parameters']
loaded_profiles = []
for raw_param in raw_params:
ps_path = raw_param['Name']
dp = custom_persistors.load_from_dict(json.loads(raw_param["Value"]))
if rename_imported_profiles:
# Update profile's name to the path on paramstore where it was fetched from = ps_path
return loaded_profiles
def saveprofiles(self, profiles, *args, **kwargs):
We could implement this using this API:
But it would require that
* the application has write permission to SSM
* there is not usually a good reason for applications to change their credentials while they are running
* this is generally the domain of a separate administrative process
* the user knows the KMS key to use for encryption and can supply that here
* makes the API kind of complicated
* the user hasn't set any weird names for parameters
* otherwise the parameters will be written all over their paramstore heirarchy, or writes may fail entirely
For these reasons, we leave this unimplemented.
raise NotImplementedError()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.