STRLCPY/tsunami-security-scanner

PiperOrigin-RevId: 464586101
Change-Id: I6edc30b83abbafcd349f23f81b645d94c7a5ddcf

John Y. Kim committed with Copybara-Service 2 years ago

9dc10b95

1 parent c298f32b

Total 2 files Show one by one

■ ■ ■ ■ ■ ■

plugin_server/py/plugin_service.py

		skipped 16 lines
17	17		from typing import cast
18	18
19	19		from absl import logging
	20	+
20	21		from tsunami.plugin_server.py import tsunami_plugin
21	22		from tsunami.proto import detection_pb2
22	23		from tsunami.proto import plugin_representation_pb2
		skipped 1 lines
24	25		from tsunami.proto import plugin_service_pb2_grpc
25	26
26	27		RunResponse = plugin_service_pb2.RunResponse
	28	+	ListPluginsRequest = plugin_service_pb2.ListPluginsRequest
	29	+	ListPluginsResponse = plugin_service_pb2.ListPluginsResponse
	30	+	_PluginServiceServicer = plugin_service_pb2_grpc.PluginServiceServicer
27	31		_PluginType = plugin_representation_pb2.PluginInfo.PluginType
28	32
29	33		_DETECTION_TIMEOUT = 60
		skipped 42 lines
72	76		response.reports.CopyFrom(report_list)
73	77		return response
74	78
	79	+	def ListPlugins(
	80	+	self, request: ListPluginsRequest,
	81	+	servicer_context: _PluginServiceServicer) -> ListPluginsResponse:
	82	+	response = ListPluginsResponse()
	83	+	response.plugins.MergeFrom(
	84	+	[plugin.GetPluginDefinition() for plugin in self.py_plugins])
	85	+	return response
	86	+

■ ■ ■ ■ ■ ■

plugin_server/py/plugin_service_test.py

		skipped 18 lines
19	19		import ipaddr
20	20
21	21		from google.protobuf import timestamp_pb2
	22	+	from net.proto2.contrib.pyutil import compare
22	23		from testing.pybase import googletest
23	24		from tsunami.plugin_server.py import plugin_service
24	25		from tsunami.plugin_server.py import tsunami_plugin
		skipped 13 lines
38	39		_ServiceDescriptor = plugin_service_pb2.DESCRIPTOR.services_by_name[
39	40		'PluginService']
40	41		_RunMethod = _ServiceDescriptor.methods_by_name['Run']
	42	+	_ListPluginsMethod = _ServiceDescriptor.methods_by_name['ListPlugins']
41	43		MAX_WORKERS = 1
42	44
43	45
		skipped 53 lines
97	99		response, _, _, _ = rpc.termination()
98	100
99	101		self.assertEmpty(response.reports.detection_reports)
	102	+
	103	+	def test_list_plugins_plugins_registered_returns_valid_response(self):
	104	+	request = plugin_service.ListPluginsRequest()
	105	+	rpc = self._server.invoke_unary_unary(_ListPluginsMethod, (), request, None)
	106	+	response, _, _, _ = rpc.termination()
	107	+	compare.assertProto2Equal(
	108	+	self,
	109	+	plugin_service.ListPluginsResponse(
	110	+	plugins=[self.test_plugin.GetPluginDefinition()]), response)
100	111
101	112
102	113		def _build_network_endpoint(ip: str, port: int) -> _NetworkEndpoint:
		skipped 57 lines