¡@
OpenStack Study: secgroup.py**** CubicPower OpenStack Study **** def mk_body(**kwargs): """Convenience function creates and dumps dictionary to string. :param kwargs: the key/value pirs to be dumped into a json string. :returns: a json string. """ return json.dumps(kwargs, ensure_ascii=False)
**** CubicPower OpenStack Study **** def query_security_profiles(cluster, fields=None, filters=None): return get_all_query_pages( _build_uri_path(SECPROF_RESOURCE, fields=fields, filters=filters), cluster)
**** CubicPower OpenStack Study **** def create_security_profile(cluster, tenant_id, neutron_id, security_profile): """Create a security profile on the NSX backend. :param cluster: a NSX cluster object reference :param tenant_id: identifier of the Neutron tenant :param neutron_id: neutron security group identifier :param security_profile: dictionary with data for configuring the NSX security profile. """ path = "/ws.v1/security-profile" # Allow all dhcp responses and all ingress traffic hidden_rules = {'logical_port_egress_rules': [{'ethertype': 'IPv4', 'protocol': constants.PROTO_NUM_UDP, 'port_range_min': constants.DHCP_RESPONSE_PORT, 'port_range_max': constants.DHCP_RESPONSE_PORT, 'ip_prefix': '0.0.0.0/0'}], 'logical_port_ingress_rules': [{'ethertype': 'IPv4'}, {'ethertype': 'IPv6'}]} display_name = utils.check_and_truncate(security_profile.get('name')) # NOTE(salv-orlando): neutron-id tags are prepended with 'q' for # historical reasons body = mk_body( tags=utils.get_tags(os_tid=tenant_id, q_sec_group_id=neutron_id), display_name=display_name, logical_port_ingress_rules=( hidden_rules['logical_port_ingress_rules']), logical_port_egress_rules=hidden_rules['logical_port_egress_rules'] ) rsp = do_request(HTTP_POST, path, body, cluster=cluster) if security_profile.get('name') == 'default': # If security group is default allow ip traffic between # members of the same security profile is allowed and ingress traffic # from the switch rules = {'logical_port_egress_rules': [{'ethertype': 'IPv4', 'profile_uuid': rsp['uuid']}, {'ethertype': 'IPv6', 'profile_uuid': rsp['uuid']}], 'logical_port_ingress_rules': [{'ethertype': 'IPv4'}, {'ethertype': 'IPv6'}]} update_security_group_rules(cluster, rsp['uuid'], rules) LOG.debug(_("Created Security Profile: %s"), rsp) return rsp
**** CubicPower OpenStack Study **** def update_security_group_rules(cluster, spid, rules): path = "/ws.v1/security-profile/%s" % spid # Allow all dhcp responses in rules['logical_port_egress_rules'].append( {'ethertype': 'IPv4', 'protocol': constants.PROTO_NUM_UDP, 'port_range_min': constants.DHCP_RESPONSE_PORT, 'port_range_max': constants.DHCP_RESPONSE_PORT, 'ip_prefix': '0.0.0.0/0'}) # If there are no ingress rules add bunk rule to drop all ingress traffic if not rules['logical_port_ingress_rules']: rules['logical_port_ingress_rules'].append( {'ethertype': 'IPv4', 'ip_prefix': '127.0.0.1/32'}) try: body = mk_body( logical_port_ingress_rules=rules['logical_port_ingress_rules'], logical_port_egress_rules=rules['logical_port_egress_rules']) rsp = do_request(HTTP_PUT, path, body, cluster=cluster) except exceptions.NotFound as e: LOG.error(format_exception("Unknown", e, locals())) #FIXME(salvatore-orlando): This should not raise NeutronException raise exceptions.NeutronException() LOG.debug(_("Updated Security Profile: %s"), rsp) return rsp
**** CubicPower OpenStack Study **** def update_security_profile(cluster, spid, name): return do_request(HTTP_PUT, _build_uri_path(SECPROF_RESOURCE, resource_id=spid), json.dumps({ "display_name": utils.check_and_truncate(name) }), cluster=cluster)
**** CubicPower OpenStack Study **** def delete_security_profile(cluster, spid): path = "/ws.v1/security-profile/%s" % spid try: do_request(HTTP_DELETE, path, cluster=cluster) except exceptions.NotFound: with excutils.save_and_reraise_exception(): # This is not necessarily an error condition LOG.warn(_("Unable to find security profile %s on NSX backend"), spid)
|