<s> """<STR_LIT>""" <EOL> from oslo_log import log as logging <EOL> from oslo_serialization import jsonutils <EOL> from webob import exc <EOL> from nova . api . openstack import common <EOL> from nova . api . openstack . compute . schemas import security_groups as schema_security_groups <EOL> from nova . api . openstack import extensions <EOL> from nova . api . openstack import wsgi <EOL> from nova import compute <EOL> from nova import exception <EOL> from nova . i18n import _ <EOL> from nova . network . security_group import openstack_driver <EOL> from nova . virt import netutils <EOL> LOG = logging . getLogger ( __name__ ) <EOL> ALIAS = '<STR_LIT>' <EOL> ATTRIBUTE_NAME = '<STR_LIT>' <EOL> authorize = extensions . os_compute_authorizer ( ALIAS ) <EOL> softauth = extensions . os_compute_soft_authorizer ( ALIAS ) <EOL> def _authorize_context ( req ) : <EOL> context = req . environ [ '<STR_LIT>' ] <EOL> authorize ( context ) <EOL> return context <EOL> class SecurityGroupControllerBase ( wsgi . Controller ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . security_group_api = ( <EOL> openstack_driver . get_openstack_security_group_driver ( <EOL> skip_policy_check = True ) ) <EOL> self . compute_api = compute . API ( <EOL> security_group_api = self . security_group_api , skip_policy_check = True ) <EOL> def _format_security_group_rule ( self , context , rule , group_rule_data = None ) : <EOL> """<STR_LIT>""" <EOL> sg_rule = { } <EOL> sg_rule [ '<STR_LIT:id>' ] = rule [ '<STR_LIT:id>' ] <EOL> sg_rule [ '<STR_LIT>' ] = rule [ '<STR_LIT>' ] <EOL> sg_rule [ '<STR_LIT>' ] = rule [ '<STR_LIT>' ] <EOL> sg_rule [ '<STR_LIT>' ] = rule [ '<STR_LIT>' ] <EOL> sg_rule [ '<STR_LIT>' ] = rule [ '<STR_LIT>' ] <EOL> sg_rule [ '<STR_LIT>' ] = { } <EOL> sg_rule [ '<STR_LIT>' ] = { } <EOL> if rule [ '<STR_LIT>' ] : <EOL> try : <EOL> source_group = self . security_group_api . get ( <EOL> context , id = rule [ '<STR_LIT>' ] ) <EOL> except exception . SecurityGroupNotFound : <EOL> LOG . debug ( "<STR_LIT>" , <EOL> rule [ '<STR_LIT>' ] ) <EOL> return <EOL> sg_rule [ '<STR_LIT>' ] = { '<STR_LIT:name>' : source_group . get ( '<STR_LIT:name>' ) , <EOL> '<STR_LIT>' : source_group . get ( '<STR_LIT>' ) } <EOL> elif group_rule_data : <EOL> sg_rule [ '<STR_LIT>' ] = group_rule_data <EOL> else : <EOL> sg_rule [ '<STR_LIT>' ] = { '<STR_LIT>' : rule [ '<STR_LIT>' ] } <EOL> return sg_rule <EOL> def _format_security_group ( self , context , group ) : <EOL> security_group = { } <EOL> security_group [ '<STR_LIT:id>' ] = group [ '<STR_LIT:id>' ] <EOL> security_group [ '<STR_LIT:description>' ] = group [ '<STR_LIT:description>' ] <EOL> security_group [ '<STR_LIT:name>' ] = group [ '<STR_LIT:name>' ] <EOL> security_group [ '<STR_LIT>' ] = group [ '<STR_LIT>' ] <EOL> security_group [ '<STR_LIT>' ] = [ ] <EOL> for rule in group [ '<STR_LIT>' ] : <EOL> formatted_rule = self . _format_security_group_rule ( context , rule ) <EOL> if formatted_rule : <EOL> security_group [ '<STR_LIT>' ] += [ formatted_rule ] <EOL> return security_group <EOL> def _from_body ( self , body , key ) : <EOL> if not body : <EOL> raise exc . HTTPBadRequest ( <EOL> explanation = _ ( "<STR_LIT>" ) ) <EOL> value = body . get ( key , None ) <EOL> if value is None : <EOL> raise exc . HTTPBadRequest ( <EOL> explanation = _ ( "<STR_LIT>" ) % key ) <EOL> return value <EOL> class SecurityGroupController ( SecurityGroupControllerBase ) : <EOL> """<STR_LIT>""" <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def show ( self , req , id ) : <EOL> """<STR_LIT>""" <EOL> context = _authorize_context ( req ) <EOL> try : <EOL> id = self . security_group_api . validate_id ( id ) <EOL> security_group = self . security_group_api . get ( context , None , id , <EOL> map_exception = True ) <EOL> except exception . SecurityGroupNotFound as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . Invalid as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> return { '<STR_LIT>' : self . _format_security_group ( context , <EOL> security_group ) } <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> @ wsgi . response ( <NUM_LIT> ) <EOL> def delete ( self , req , id ) : <EOL> """<STR_LIT>""" <EOL> context = _authorize_context ( req ) <EOL> try : <EOL> id = self . security_group_api . validate_id ( id ) <EOL> security_group = self . security_group_api . get ( context , None , id , <EOL> map_exception = True ) <EOL> self . security_group_api . destroy ( context , security_group ) <EOL> except exception . SecurityGroupNotFound as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . Invalid as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> @ extensions . expected_errors ( <NUM_LIT> ) <EOL> def index ( self , req ) : <EOL> """<STR_LIT>""" <EOL> context = _authorize_context ( req ) <EOL> search_opts = { } <EOL> search_opts . update ( req . GET ) <EOL> project_id = context . project_id <EOL> raw_groups = self . security_group_api . list ( context , <EOL> project = project_id , <EOL> search_opts = search_opts ) <EOL> limited_list = common . limited ( raw_groups , req ) <EOL> result = [ self . _format_security_group ( context , group ) <EOL> for group in limited_list ] <EOL> return { '<STR_LIT>' : <EOL> list ( sorted ( result , <EOL> key = lambda k : ( k [ '<STR_LIT>' ] , k [ '<STR_LIT:name>' ] ) ) ) } <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def create ( self , req , body ) : <EOL> """<STR_LIT>""" <EOL> context = _authorize_context ( req ) <EOL> security_group = self . _from_body ( body , '<STR_LIT>' ) <EOL> group_name = security_group . get ( '<STR_LIT:name>' , None ) <EOL> group_description = security_group . get ( '<STR_LIT:description>' , None ) <EOL> try : <EOL> self . security_group_api . validate_property ( group_name , '<STR_LIT:name>' , None ) <EOL> self . security_group_api . validate_property ( group_description , <EOL> '<STR_LIT:description>' , None ) <EOL> group_ref = self . security_group_api . create_security_group ( <EOL> context , group_name , group_description ) <EOL> except exception . Invalid as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> except exception . SecurityGroupLimitExceeded as exp : <EOL> raise exc . HTTPForbidden ( explanation = exp . format_message ( ) ) <EOL> return { '<STR_LIT>' : self . _format_security_group ( context , <EOL> group_ref ) } <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def update ( self , req , id , body ) : <EOL> """<STR_LIT>""" <EOL> context = _authorize_context ( req ) <EOL> try : <EOL> id = self . security_group_api . validate_id ( id ) <EOL> security_group = self . security_group_api . get ( context , None , id , <EOL> map_exception = True ) <EOL> except exception . SecurityGroupNotFound as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . Invalid as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> security_group_data = self . _from_body ( body , '<STR_LIT>' ) <EOL> group_name = security_group_data . get ( '<STR_LIT:name>' , None ) <EOL> group_description = security_group_data . get ( '<STR_LIT:description>' , None ) <EOL> try : <EOL> self . security_group_api . validate_property ( group_name , '<STR_LIT:name>' , None ) <EOL> self . security_group_api . validate_property ( group_description , <EOL> '<STR_LIT:description>' , None ) <EOL> group_ref = self . security_group_api . update_security_group ( <EOL> context , security_group , group_name , group_description ) <EOL> except exception . SecurityGroupNotFound as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . Invalid as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> return { '<STR_LIT>' : self . _format_security_group ( context , <EOL> group_ref ) } <EOL> class SecurityGroupRulesController ( SecurityGroupControllerBase ) : <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) ) <EOL> def create ( self , req , body ) : <EOL> context = _authorize_context ( req ) <EOL> sg_rule = self . _from_body ( body , '<STR_LIT>' ) <EOL> try : <EOL> parent_group_id = self . security_group_api . validate_id ( <EOL> sg_rule . get ( '<STR_LIT>' ) ) <EOL> security_group = self . security_group_api . get ( context , None , <EOL> parent_group_id , <EOL> map_exception = True ) <EOL> new_rule = self . _rule_args_to_dict ( context , <EOL> to_port = sg_rule . get ( '<STR_LIT>' ) , <EOL> from_port = sg_rule . get ( '<STR_LIT>' ) , <EOL> ip_protocol = sg_rule . get ( '<STR_LIT>' ) , <EOL> cidr = sg_rule . get ( '<STR_LIT>' ) , <EOL> group_id = sg_rule . get ( '<STR_LIT>' ) ) <EOL> except ( exception . Invalid , exception . InvalidCidr ) as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> except exception . SecurityGroupNotFound as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> if new_rule is None : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> raise exc . HTTPBadRequest ( explanation = msg ) <EOL> new_rule [ '<STR_LIT>' ] = security_group [ '<STR_LIT:id>' ] <EOL> if '<STR_LIT>' in new_rule : <EOL> net , prefixlen = netutils . get_net_and_prefixlen ( new_rule [ '<STR_LIT>' ] ) <EOL> if net not in ( '<STR_LIT>' , '<STR_LIT>' ) and prefixlen == '<STR_LIT:0>' : <EOL> msg = _ ( "<STR_LIT>" ) % new_rule [ '<STR_LIT>' ] <EOL> raise exc . HTTPBadRequest ( explanation = msg ) <EOL> group_rule_data = None <EOL> try : <EOL> if sg_rule . get ( '<STR_LIT>' ) : <EOL> source_group = self . security_group_api . get ( <EOL> context , id = sg_rule [ '<STR_LIT>' ] ) <EOL> group_rule_data = { '<STR_LIT:name>' : source_group . get ( '<STR_LIT:name>' ) , <EOL> '<STR_LIT>' : source_group . get ( '<STR_LIT>' ) } <EOL> security_group_rule = ( <EOL> self . security_group_api . create_security_group_rule ( <EOL> context , security_group , new_rule ) ) <EOL> except exception . Invalid as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> except exception . SecurityGroupNotFound as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . SecurityGroupLimitExceeded as exp : <EOL> raise exc . HTTPForbidden ( explanation = exp . format_message ( ) ) <EOL> formatted_rule = self . _format_security_group_rule ( context , <EOL> security_group_rule , <EOL> group_rule_data ) <EOL> return { "<STR_LIT>" : formatted_rule } <EOL> def _rule_args_to_dict ( self , context , to_port = None , from_port = None , <EOL> ip_protocol = None , cidr = None , group_id = None ) : <EOL> if group_id is not None : <EOL> group_id = self . security_group_api . validate_id ( group_id ) <EOL> self . security_group_api . get ( context , id = group_id ) <EOL> return self . security_group_api . new_group_ingress_rule ( <EOL> group_id , ip_protocol , from_port , to_port ) <EOL> else : <EOL> cidr = self . security_group_api . parse_cidr ( cidr ) <EOL> return self . security_group_api . new_cidr_ingress_rule ( <EOL> cidr , ip_protocol , from_port , to_port ) <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) ) <EOL> @ wsgi . response ( <NUM_LIT> ) <EOL> def delete ( self , req , id ) : <EOL> context = _authorize_context ( req ) <EOL> try : <EOL> id = self . security_group_api . validate_id ( id ) <EOL> rule = self . security_group_api . get_rule ( context , id ) <EOL> group_id = rule [ '<STR_LIT>' ] <EOL> security_group = self . security_group_api . get ( context , None , <EOL> group_id , <EOL> map_exception = True ) <EOL> self . security_group_api . remove_rules ( context , security_group , <EOL> [ rule [ '<STR_LIT:id>' ] ] ) <EOL> except exception . SecurityGroupNotFound as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . NoUniqueMatch as exp : <EOL> raise exc . HTTPConflict ( explanation = exp . format_message ( ) ) <EOL> except exception . Invalid as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> class ServerSecurityGroupController ( SecurityGroupControllerBase ) : <EOL> @ extensions . expected_errors ( <NUM_LIT> ) <EOL> def index ( self , req , server_id ) : <EOL> """<STR_LIT>""" <EOL> context = _authorize_context ( req ) <EOL> self . security_group_api . ensure_default ( context ) <EOL> try : <EOL> instance = common . get_instance ( self . compute_api , context , <EOL> server_id ) <EOL> groups = self . security_group_api . get_instance_security_groups ( <EOL> context , instance , True ) <EOL> except ( exception . SecurityGroupNotFound , <EOL> exception . InstanceNotFound ) as exp : <EOL> msg = exp . format_message ( ) <EOL> raise exc . HTTPNotFound ( explanation = msg ) <EOL> result = [ self . _format_security_group ( context , group ) <EOL> for group in groups ] <EOL> return { '<STR_LIT>' : <EOL> list ( sorted ( result , <EOL> key = lambda k : ( k [ '<STR_LIT>' ] , k [ '<STR_LIT:name>' ] ) ) ) } <EOL> class SecurityGroupActionController ( wsgi . Controller ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( SecurityGroupActionController , self ) . __init__ ( * args , ** kwargs ) <EOL> self . security_group_api = ( <EOL> openstack_driver . get_openstack_security_group_driver ( <EOL> skip_policy_check = True ) ) <EOL> self . compute_api = compute . API ( <EOL> security_group_api = self . security_group_api , skip_policy_check = True ) <EOL> def _parse ( self , body , action ) : <EOL> try : <EOL> body = body [ action ] <EOL> group_name = body [ '<STR_LIT:name>' ] <EOL> except TypeError : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> raise exc . HTTPBadRequest ( explanation = msg ) <EOL> except KeyError : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> raise exc . HTTPBadRequest ( explanation = msg ) <EOL> if not group_name or group_name . strip ( ) == '<STR_LIT>' : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> raise exc . HTTPBadRequest ( explanation = msg ) <EOL> return group_name <EOL> def _invoke ( self , method , context , id , group_name ) : <EOL> instance = common . get_instance ( self . compute_api , context , id ) <EOL> method ( context , instance , group_name ) <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) ) <EOL> @ wsgi . response ( <NUM_LIT> ) <EOL> @ wsgi . action ( '<STR_LIT>' ) <EOL> def _addSecurityGroup ( self , req , id , body ) : <EOL> context = req . environ [ '<STR_LIT>' ] <EOL> authorize ( context ) <EOL> group_name = self . _parse ( body , '<STR_LIT>' ) <EOL> try : <EOL> return self . _invoke ( self . security_group_api . add_to_instance , <EOL> context , id , group_name ) <EOL> except ( exception . SecurityGroupNotFound , <EOL> exception . InstanceNotFound ) as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . NoUniqueMatch as exp : <EOL> raise exc . HTTPConflict ( explanation = exp . format_message ( ) ) <EOL> except ( exception . SecurityGroupCannotBeApplied , <EOL> exception . SecurityGroupExistsForInstance ) as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> @ extensions . expected_errors ( ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) ) <EOL> @ wsgi . response ( <NUM_LIT> ) <EOL> @ wsgi . action ( '<STR_LIT>' ) <EOL> def _removeSecurityGroup ( self , req , id , body ) : <EOL> context = req . environ [ '<STR_LIT>' ] <EOL> authorize ( context ) <EOL> group_name = self . _parse ( body , '<STR_LIT>' ) <EOL> try : <EOL> return self . _invoke ( self . security_group_api . remove_from_instance , <EOL> context , id , group_name ) <EOL> except ( exception . SecurityGroupNotFound , <EOL> exception . InstanceNotFound ) as exp : <EOL> raise exc . HTTPNotFound ( explanation = exp . format_message ( ) ) <EOL> except exception . NoUniqueMatch as exp : <EOL> raise exc . HTTPConflict ( explanation = exp . format_message ( ) ) <EOL> except exception . SecurityGroupNotExistsForInstance as exp : <EOL> raise exc . HTTPBadRequest ( explanation = exp . format_message ( ) ) <EOL> class SecurityGroupsOutputController ( wsgi . Controller ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( SecurityGroupsOutputController , self ) . __init__ ( * args , ** kwargs ) <EOL> self . compute_api = compute . API ( skip_policy_check = True ) <EOL> self . security_group_api = ( <EOL> openstack_driver . get_openstack_security_group_driver ( <EOL> skip_policy_check = True ) ) <EOL> def _extend_servers ( self , req , servers ) : <EOL> if not len ( servers ) : <EOL> return <EOL> key = "<STR_LIT>" <EOL> context = req . environ [ '<STR_LIT>' ] <EOL> if not softauth ( context ) : <EOL> return <EOL> if not openstack_driver . is_neutron_security_groups ( ) : <EOL> for server in servers : <EOL> instance = req . get_db_instance ( server [ '<STR_LIT:id>' ] ) <EOL> groups = instance . get ( key ) <EOL> if groups : <EOL> server [ ATTRIBUTE_NAME ] = [ { "<STR_LIT:name>" : group [ "<STR_LIT:name>" ] } <EOL> for group in groups ] <EOL> else : <EOL> if req . method != '<STR_LIT:POST>' : <EOL> sg_instance_bindings = ( <EOL> self . security_group_api <EOL> . get_instances_security_groups_bindings ( context , <EOL> servers ) ) <EOL> for server in servers : <EOL> groups = sg_instance_bindings . get ( server [ '<STR_LIT:id>' ] ) <EOL> if groups : <EOL> server [ ATTRIBUTE_NAME ] = groups <EOL> else : <EOL> req_obj = jsonutils . loads ( req . body ) <EOL> servers [ <NUM_LIT:0> ] [ ATTRIBUTE_NAME ] = req_obj [ '<STR_LIT>' ] . get ( <EOL> ATTRIBUTE_NAME , [ { '<STR_LIT:name>' : '<STR_LIT:default>' } ] ) <EOL> def _show ( self , req , resp_obj ) : <EOL> if '<STR_LIT>' in resp_obj . obj : <EOL> self . _extend_servers ( req , [ resp_obj . obj [ '<STR_LIT>' ] ] ) <EOL> @ wsgi . extends <EOL> def show ( self , req , resp_obj , id ) : <EOL> return self . _show ( req , resp_obj ) <EOL> @ wsgi . extends <EOL> def create ( self , req , resp_obj , body ) : <EOL> return self . _show ( req , resp_obj ) <EOL> @ wsgi . extends <EOL> def detail ( self , req , resp_obj ) : <EOL> self . _extend_servers ( req , list ( resp_obj . obj [ '<STR_LIT>' ] ) ) <EOL> class SecurityGroups ( extensions . V21APIExtensionBase ) : <EOL> """<STR_LIT>""" <EOL> name = "<STR_LIT>" <EOL> alias = ALIAS <EOL> version = <NUM_LIT:1> <EOL> def get_controller_extensions ( self ) : <EOL> secgrp_output_ext = extensions . ControllerExtension ( <EOL> self , '<STR_LIT>' , SecurityGroupsOutputController ( ) ) <EOL> secgrp_act_ext = extensions . ControllerExtension ( <EOL> self , '<STR_LIT>' , SecurityGroupActionController ( ) ) <EOL> return [ secgrp_output_ext , secgrp_act_ext ] <EOL> def get_resources ( self ) : <EOL> secgrp_ext = extensions . ResourceExtension ( ALIAS , <EOL> SecurityGroupController ( ) ) <EOL> server_secgrp_ext = extensions . ResourceExtension ( <EOL> ALIAS , <EOL> controller = ServerSecurityGroupController ( ) , <EOL> parent = dict ( member_name = '<STR_LIT>' , collection_name = '<STR_LIT>' ) ) <EOL> secgrp_rules_ext = extensions . ResourceExtension ( <EOL> '<STR_LIT>' , <EOL> controller = SecurityGroupRulesController ( ) ) <EOL> return [ secgrp_ext , server_secgrp_ext , secgrp_rules_ext ] <EOL> def server_create ( self , server_dict , create_kwargs , body_deprecated_param ) : <EOL> security_groups = server_dict . get ( ATTRIBUTE_NAME ) <EOL> if security_groups is not None : <EOL> create_kwargs [ '<STR_LIT>' ] = [ <EOL> sg [ '<STR_LIT:name>' ] for sg in security_groups if sg . get ( '<STR_LIT:name>' ) ] <EOL> create_kwargs [ '<STR_LIT>' ] = list ( <EOL> set ( create_kwargs [ '<STR_LIT>' ] ) ) <EOL> def get_server_create_schema ( self , version ) : <EOL> if version == '<STR_LIT>' : <EOL> return schema_security_groups . server_create_v20 <EOL> return schema_security_groups . server_create </s>
