How to use ssm method in localstack

Best Python code snippet using localstack_python

CISCO-SSM-PROV-MIB.py

Source:CISCO-SSM-PROV-MIB.py

1#2# PySNMP MIB module CISCO-SSM-PROV-MIB (http://snmplabs.com/pysmi)3# ASN.1 source file:///Users/davwang4/Dev/mibs.snmplabs.com/asn1/CISCO-SSM-PROV-MIB4# Produced by pysmi-0.3.4 at Mon Apr 29 17:56:15 20195# On host DAVWANG4-M-1475 platform Darwin version 18.5.0 by user davwang46# Using Python version 3.7.3 (default, Mar 27 2019, 09:23:15) 7#8ObjectIdentifier, OctetString, Integer = mibBuilder.importSymbols("ASN1", "ObjectIdentifier", "OctetString", "Integer")9NamedValues, = mibBuilder.importSymbols("ASN1-ENUMERATION", "NamedValues")10ConstraintsUnion, SingleValueConstraint, ConstraintsIntersection, ValueRangeConstraint, ValueSizeConstraint = mibBuilder.importSymbols("ASN1-REFINEMENT", "ConstraintsUnion", "SingleValueConstraint", "ConstraintsIntersection", "ValueRangeConstraint", "ValueSizeConstraint")11ciscoMgmt, = mibBuilder.importSymbols("CISCO-SMI", "ciscoMgmt")12entPhysicalIndex, = mibBuilder.importSymbols("ENTITY-MIB", "entPhysicalIndex")13InterfaceIndex, = mibBuilder.importSymbols("IF-MIB", "InterfaceIndex")14SnmpAdminString, = mibBuilder.importSymbols("SNMP-FRAMEWORK-MIB", "SnmpAdminString")15ModuleCompliance, NotificationGroup, ObjectGroup = mibBuilder.importSymbols("SNMPv2-CONF", "ModuleCompliance", "NotificationGroup", "ObjectGroup")16ModuleIdentity, MibIdentifier, Counter64, IpAddress, Integer32, ObjectIdentity, Bits, TimeTicks, Gauge32, NotificationType, Unsigned32, Counter32, iso, MibScalar, MibTable, MibTableRow, MibTableColumn = mibBuilder.importSymbols("SNMPv2-SMI", "ModuleIdentity", "MibIdentifier", "Counter64", "IpAddress", "Integer32", "ObjectIdentity", "Bits", "TimeTicks", "Gauge32", "NotificationType", "Unsigned32", "Counter32", "iso", "MibScalar", "MibTable", "MibTableRow", "MibTableColumn")17DisplayString, TruthValue, TextualConvention, RowStatus = mibBuilder.importSymbols("SNMPv2-TC", "DisplayString", "TruthValue", "TextualConvention", "RowStatus")18ciscoSsmProvMIB = ModuleIdentity((1, 3, 6, 1, 4, 1, 9, 9, 448))19ciscoSsmProvMIB.setRevisions(('2005-02-15 00:00',))20if mibBuilder.loadTexts: ciscoSsmProvMIB.setLastUpdated('200502150000Z')21if mibBuilder.loadTexts: ciscoSsmProvMIB.setOrganization('Cisco Systems Inc.')22ciscoSsmProvMIBObjects = MibIdentifier((1, 3, 6, 1, 4, 1, 9, 9, 448, 1))23ciscoSsmProvMIBConform = MibIdentifier((1, 3, 6, 1, 4, 1, 9, 9, 448, 2))24ssmProvConfiguration = MibIdentifier((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1))25ssmProvFeatureTable = MibTable((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 1), )26if mibBuilder.loadTexts: ssmProvFeatureTable.setStatus('current')27ssmProvFeatureEntry = MibTableRow((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 1, 1), ).setIndexNames((0, "ENTITY-MIB", "entPhysicalIndex"), (0, "CISCO-SSM-PROV-MIB", "ssmProvFeatureName"))28if mibBuilder.loadTexts: ssmProvFeatureEntry.setStatus('current')29ssmProvFeatureName = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 1, 1, 1), SnmpAdminString().subtype(subtypeSpec=ValueSizeConstraint(1, 80)))30if mibBuilder.loadTexts: ssmProvFeatureName.setStatus('current')31ssmProvFeatureNeedsImage = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 1, 1, 2), TruthValue()).setMaxAccess("readonly")32if mibBuilder.loadTexts: ssmProvFeatureNeedsImage.setStatus('current')33ssmProvFeatureIfTable = MibTable((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2), )34if mibBuilder.loadTexts: ssmProvFeatureIfTable.setStatus('current')35ssmProvFeatureIfEntry = MibTableRow((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2, 1), ).setIndexNames((0, "CISCO-SSM-PROV-MIB", "ssmProvFeatureIfStartPort"), (0, "CISCO-SSM-PROV-MIB", "ssmProvFeatureIfEndPort"), (0, "CISCO-SSM-PROV-MIB", "ssmProvFeatureIfFeatureName"))36if mibBuilder.loadTexts: ssmProvFeatureIfEntry.setStatus('current')37ssmProvFeatureIfStartPort = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2, 1, 1), InterfaceIndex())38if mibBuilder.loadTexts: ssmProvFeatureIfStartPort.setStatus('current')39ssmProvFeatureIfEndPort = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2, 1, 2), InterfaceIndex())40if mibBuilder.loadTexts: ssmProvFeatureIfEndPort.setStatus('current')41ssmProvFeatureIfFeatureName = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2, 1, 3), SnmpAdminString().subtype(subtypeSpec=ValueSizeConstraint(1, 80)))42if mibBuilder.loadTexts: ssmProvFeatureIfFeatureName.setStatus('current')43ssmProvFeatureIfForceRemove = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2, 1, 4), Integer32().subtype(subtypeSpec=ConstraintsUnion(SingleValueConstraint(1, 2))).clone(namedValues=NamedValues(("forceReset", 1), ("noop", 2))).clone('noop')).setMaxAccess("readcreate")44if mibBuilder.loadTexts: ssmProvFeatureIfForceRemove.setStatus('current')45ssmProvFeatureIfPartnerImageURI = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2, 1, 5), SnmpAdminString().subtype(subtypeSpec=ValueSizeConstraint(1, 80))).setMaxAccess("readcreate")46if mibBuilder.loadTexts: ssmProvFeatureIfPartnerImageURI.setStatus('current')47ssmProvFeatureIfRowStatus = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 2, 1, 6), RowStatus()).setMaxAccess("readcreate")48if mibBuilder.loadTexts: ssmProvFeatureIfRowStatus.setStatus('current')49ssmProvDppTable = MibTable((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 3), )50if mibBuilder.loadTexts: ssmProvDppTable.setStatus('current')51ssmProvDppEntry = MibTableRow((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 3, 1), ).setIndexNames((0, "CISCO-SSM-PROV-MIB", "ssmProvDppStartPort"))52if mibBuilder.loadTexts: ssmProvDppEntry.setStatus('current')53ssmProvDppStartPort = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 3, 1, 1), InterfaceIndex())54if mibBuilder.loadTexts: ssmProvDppStartPort.setStatus('current')55ssmProvDppEndPort = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 3, 1, 2), InterfaceIndex()).setMaxAccess("readonly")56if mibBuilder.loadTexts: ssmProvDppEndPort.setStatus('current')57ssmProvDppName = MibTableColumn((1, 3, 6, 1, 4, 1, 9, 9, 448, 1, 1, 3, 1, 3), SnmpAdminString().subtype(subtypeSpec=ValueSizeConstraint(1, 80))).setMaxAccess("readonly")58if mibBuilder.loadTexts: ssmProvDppName.setStatus('current')59ssmProvMIBCompliances = MibIdentifier((1, 3, 6, 1, 4, 1, 9, 9, 448, 2, 1))60ssmProvMIBGroups = MibIdentifier((1, 3, 6, 1, 4, 1, 9, 9, 448, 2, 2))61ssmProvMIBCompliance = ModuleCompliance((1, 3, 6, 1, 4, 1, 9, 9, 448, 2, 1, 1)).setObjects(("CISCO-SSM-PROV-MIB", "ssmProvFeatureGroup"), ("CISCO-SSM-PROV-MIB", "ssmProvFeatureIfGroup"), ("CISCO-SSM-PROV-MIB", "ssmProvDppGroup"))62if getattr(mibBuilder, 'version', (0, 0, 0)) > (4, 4, 0):63    ssmProvMIBCompliance = ssmProvMIBCompliance.setStatus('current')64ssmProvFeatureGroup = ObjectGroup((1, 3, 6, 1, 4, 1, 9, 9, 448, 2, 2, 1)).setObjects(("CISCO-SSM-PROV-MIB", "ssmProvFeatureNeedsImage"))65if getattr(mibBuilder, 'version', (0, 0, 0)) > (4, 4, 0):66    ssmProvFeatureGroup = ssmProvFeatureGroup.setStatus('current')67ssmProvFeatureIfGroup = ObjectGroup((1, 3, 6, 1, 4, 1, 9, 9, 448, 2, 2, 2)).setObjects(("CISCO-SSM-PROV-MIB", "ssmProvFeatureIfForceRemove"), ("CISCO-SSM-PROV-MIB", "ssmProvFeatureIfPartnerImageURI"), ("CISCO-SSM-PROV-MIB", "ssmProvFeatureIfRowStatus"))68if getattr(mibBuilder, 'version', (0, 0, 0)) > (4, 4, 0):69    ssmProvFeatureIfGroup = ssmProvFeatureIfGroup.setStatus('current')70ssmProvDppGroup = ObjectGroup((1, 3, 6, 1, 4, 1, 9, 9, 448, 2, 2, 3)).setObjects(("CISCO-SSM-PROV-MIB", "ssmProvDppEndPort"), ("CISCO-SSM-PROV-MIB", "ssmProvDppName"))71if getattr(mibBuilder, 'version', (0, 0, 0)) > (4, 4, 0):72    ssmProvDppGroup = ssmProvDppGroup.setStatus('current')...

__init__.py

Source:__init__.py

1from operator import attrgetter2import pyangbind.lib.xpathhelper as xpathhelper3from pyangbind.lib.yangtypes import RestrictedPrecisionDecimalType, RestrictedClassType, TypedListType4from pyangbind.lib.yangtypes import YANGBool, YANGListType, YANGDynClass, ReferenceType5from pyangbind.lib.base import PybindBase6from decimal import Decimal7from bitarray import bitarray8import __builtin__9import ssm_maps10class multicast_ssm_mapping(PybindBase):11  """12  This class was auto-generated by the PythonClass plugin for PYANG13  from YANG module brocade-mc-hms-operational - based on the path /igmp-snooping-state/multicast-ssm-mapping. Each member element of14  the container is represented as a class variable - with a specific15  YANG type.16  YANG Description: Source Specific Multicast Operational Information17  """18  __slots__ = ('_pybind_generated_by', '_path_helper', '_yang_name', '_rest_name', '_extmethods', '__ssm_maps',)19  _yang_name = 'multicast-ssm-mapping'20  _rest_name = 'multicast-ssm-mapping'21  _pybind_generated_by = 'container'22  def __init__(self, *args, **kwargs):23    path_helper_ = kwargs.pop("path_helper", None)24    if path_helper_ is False:25      self._path_helper = False26    elif path_helper_ is not None and isinstance(path_helper_, xpathhelper.YANGPathHelper):27      self._path_helper = path_helper_28    elif hasattr(self, "_parent"):29      path_helper_ = getattr(self._parent, "_path_helper", False)30      self._path_helper = path_helper_31    else:32      self._path_helper = False33    extmethods = kwargs.pop("extmethods", None)34    if extmethods is False:35      self._extmethods = False36    elif extmethods is not None and isinstance(extmethods, dict):37      self._extmethods = extmethods38    elif hasattr(self, "_parent"):39      extmethods = getattr(self._parent, "_extmethods", None)40      self._extmethods = extmethods41    else:42      self._extmethods = False43    self.__ssm_maps = YANGDynClass(base=YANGListType("prefix_list_name",ssm_maps.ssm_maps, yang_name="ssm-maps", rest_name="ssm-maps", parent=self, is_container='list', user_ordered=False, path_helper=self._path_helper, yang_keys='prefix-list-name', extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}), is_container='list', yang_name="ssm-maps", rest_name="ssm-maps", parent=self, path_helper=self._path_helper, extmethods=self._extmethods, register_paths=True, extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}, namespace='urn:brocade.com:mgmt:brocade-mc-hms-operational', defining_module='brocade-mc-hms-operational', yang_type='list', is_config=False)44    load = kwargs.pop("load", None)45    if args:46      if len(args) > 1:47        raise TypeError("cannot create a YANG container with >1 argument")48      all_attr = True49      for e in self._pyangbind_elements:50        if not hasattr(args[0], e):51          all_attr = False52          break53      if not all_attr:54        raise ValueError("Supplied object did not have the correct attributes")55      for e in self._pyangbind_elements:56        nobj = getattr(args[0], e)57        if nobj._changed() is False:58          continue59        setmethod = getattr(self, "_set_%s" % e)60        if load is None:61          setmethod(getattr(args[0], e))62        else:63          setmethod(getattr(args[0], e), load=load)64  def _path(self):65    if hasattr(self, "_parent"):66      return self._parent._path()+[self._yang_name]67    else:68      return [u'igmp-snooping-state', u'multicast-ssm-mapping']69  def _rest_path(self):70    if hasattr(self, "_parent"):71      if self._rest_name:72        return self._parent._rest_path()+[self._rest_name]73      else:74        return self._parent._rest_path()75    else:76      return [u'igmp-snooping-state', u'multicast-ssm-mapping']77  def _get_ssm_maps(self):78    """79    Getter method for ssm_maps, mapped from YANG variable /igmp_snooping_state/multicast_ssm_mapping/ssm_maps (list)80    YANG Description: Source Specific Multicast Operational Information81    """82    return self.__ssm_maps83      84  def _set_ssm_maps(self, v, load=False):85    """86    Setter method for ssm_maps, mapped from YANG variable /igmp_snooping_state/multicast_ssm_mapping/ssm_maps (list)87    If this variable is read-only (config: false) in the88    source YANG file, then _set_ssm_maps is considered as a private89    method. Backends looking to populate this variable should90    do so via calling thisObj._set_ssm_maps() directly.91    YANG Description: Source Specific Multicast Operational Information92    """93    if hasattr(v, "_utype"):94      v = v._utype(v)95    try:96      t = YANGDynClass(v,base=YANGListType("prefix_list_name",ssm_maps.ssm_maps, yang_name="ssm-maps", rest_name="ssm-maps", parent=self, is_container='list', user_ordered=False, path_helper=self._path_helper, yang_keys='prefix-list-name', extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}), is_container='list', yang_name="ssm-maps", rest_name="ssm-maps", parent=self, path_helper=self._path_helper, extmethods=self._extmethods, register_paths=True, extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}, namespace='urn:brocade.com:mgmt:brocade-mc-hms-operational', defining_module='brocade-mc-hms-operational', yang_type='list', is_config=False)97    except (TypeError, ValueError):98      raise ValueError({99          'error-string': """ssm_maps must be of a type compatible with list""",100          'defined-type': "list",101          'generated-type': """YANGDynClass(base=YANGListType("prefix_list_name",ssm_maps.ssm_maps, yang_name="ssm-maps", rest_name="ssm-maps", parent=self, is_container='list', user_ordered=False, path_helper=self._path_helper, yang_keys='prefix-list-name', extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}), is_container='list', yang_name="ssm-maps", rest_name="ssm-maps", parent=self, path_helper=self._path_helper, extmethods=self._extmethods, register_paths=True, extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}, namespace='urn:brocade.com:mgmt:brocade-mc-hms-operational', defining_module='brocade-mc-hms-operational', yang_type='list', is_config=False)""",102        })103    self.__ssm_maps = t104    if hasattr(self, '_set'):105      self._set()106  def _unset_ssm_maps(self):107    self.__ssm_maps = YANGDynClass(base=YANGListType("prefix_list_name",ssm_maps.ssm_maps, yang_name="ssm-maps", rest_name="ssm-maps", parent=self, is_container='list', user_ordered=False, path_helper=self._path_helper, yang_keys='prefix-list-name', extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}), is_container='list', yang_name="ssm-maps", rest_name="ssm-maps", parent=self, path_helper=self._path_helper, extmethods=self._extmethods, register_paths=True, extensions={u'tailf-common': {u'callpoint': u'mc-hms-ssm-map', u'cli-suppress-show-path': None}}, namespace='urn:brocade.com:mgmt:brocade-mc-hms-operational', defining_module='brocade-mc-hms-operational', yang_type='list', is_config=False)108  ssm_maps = __builtin__.property(_get_ssm_maps)...

ssm.py

Source:ssm.py

1import json2import os3import ast4import logging as log5import boto36from kinappserver import config7from kinappserver.utils import InternalError8def get_ssm_value(keyname):9    value = get_ssm_parameter(keyname, config.KMS_KEY_AWS_REGION)10    if not value:11        print('could not retrieve value %s from ssm' % value)12        return None13    else:14        return value15def get_encrpytion_creds():16    """returns the encrpytion/decryption key and iv from ssm"""17    env = os.environ.get('ENV', 'test')18    encryption_key = get_ssm_parameter('/config/' + env + '/encryption/key', config.KMS_KEY_AWS_REGION)19    iv = bytes.fromhex(get_ssm_parameter('/config/' + env + '/encryption/iv', config.KMS_KEY_AWS_REGION)) # the iv is encoded into hex()20    if not encryption_key:21        log.error('cant get encryption_creds')22        raise InternalError('cant get encryption_creds')23    return encryption_key, iv24def get_truex_creds(force_prod=False):25    env = os.environ.get('ENV', 'test')26    if force_prod:27        print('forcing prod truex creds')28        env = 'prod'29    partner_hash = get_ssm_parameter('/config/' + env + '/truex/partner_hash', config.KMS_KEY_AWS_REGION)30    callback_secret = get_ssm_parameter('/config/' + env + '/truex/callback_secret', config.KMS_KEY_AWS_REGION)31    app_id = get_ssm_parameter('/config/' + env + '/truex/app_id', config.KMS_KEY_AWS_REGION)32    if None in (app_id, partner_hash, callback_secret):33        log.error('cant get truex creds')34        raise InternalError('cant get truex creds')35    return app_id, partner_hash, callback_secret36def get_stellar_credentials():37    # get credentials from ssm. the base_seed is required, the channel-seeds are optional38    env = os.environ.get('ENV', 'test')39    account_sid = os.environ.get('STELLAR_ACCOUNT_SID', None)40    41    if env.upper()== 'TEST':42        account_sid = '0'  # for tests, always use 043    base_seed = get_ssm_parameter('/config/' + env + '/stellar/account_sid_%s/base-seed' % account_sid, config.KMS_KEY_AWS_REGION)44    channel_seeds = get_ssm_parameter('/config/' + env + '/stellar/account_sid_%s/channel-seeds' % account_sid, config.KMS_KEY_AWS_REGION)45    if base_seed is None:46        log.error('cant get base_seed, aborting')47        return None, None48    if not channel_seeds:49        return base_seed, []50    return base_seed, convert_string_to_string_array(channel_seeds)51def write_service_account():52    """"creates a service-account file if one does not exist already"""53    import os54    if os.path.exists(config.FIREBASE_SERVICE_ACCOUNT_FILE):55        return config.FIREBASE_SERVICE_ACCOUNT_FILE56    else:57        env = os.environ.get('ENV', 'test')58        service_account_json = get_ssm_parameter('/config/' + env + '/firebase/service-account', config.KMS_KEY_AWS_REGION)59        # travis is a special case - has a unique path:60        is_travis = os.environ.get('TRAVIS', 0)61        path = config.FIREBASE_SERVICE_ACCOUNT_FILE if not is_travis else '/home/travis/build/kinecosystem/kin-app-server/kinappserver/service-account.json'62        with open(path, 'w+') as the_file:63            the_file.write(service_account_json)64        return path65def convert_string_to_string_array(input):66    """converts the input from ssm to a python string array67    used for the channel seeds, which sign txs68    """69    # to add new channels, use this aws command (just plug in the secret keys and kms key70    #  aws ssm put-parameter --name /config/prod/stellar/channel-seeds --value "\"secret_key1\",\"secret_key2\",\"secret_key3\",\"secret_key4\",\"secret_key5\",\"secret_key6\"" --type SecureString --key-id <kms arn> --overwrite71    # this is used to convert the decrypted seed channels string to an array72    return json.loads('[' + str(input) + ']')73def get_ssm_parameter(param_name, kms_key_region):74    """retreives an encrpyetd value from AWSs ssm or None"""75    try:76        ssm_client = boto3.client('ssm', region_name=kms_key_region)77        print('getting param from ssm: %s' % param_name)78        res = ssm_client.get_parameter(Name=param_name, WithDecryption=True)79        return res['Parameter']['Value']80    except Exception as e:81        log.error('cant get secure value: %s from ssm' % param_name)82        print(e)83        return None84def get_security_passwords():85    """returns the kinit security password from ssm"""86    passwords = []87    env = os.environ.get('ENV', 'test')88    password = get_ssm_parameter('/config/' + env + '/misc/password', config.KMS_KEY_AWS_REGION)89    password2 = get_ssm_parameter('/config/' + env + '/misc/password2', config.KMS_KEY_AWS_REGION)90    for item in [password, password2]:91        if item:92            passwords.append(item)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.