1"""
2This module provides REST API handlers for scenario events.
3"""
4import warnings
5import falcon
6
7from .handlers import ScenarioBaseHandler
8from ...simulation import Leakage, SensorFault, SensorReadingAttack
9from ...simulation.events import SpeciesInjectionEvent
10
11
[docs]
12class ScenarioLeakageHandler(ScenarioBaseHandler):
13 """
14 Class for handling GET and POST requests concerning leakages.
15 """
[docs]
16 def on_get(self, _, resp: falcon.Response, scenario_id: str) -> None:
17 """
18 Gets all leakages of a given scenario.
19
20 Parameters
21 ----------
22 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
23 Response instance.
24 scenario_id : `str`
25 UUID of the scenario.
26 """
27 try:
28 if self.scenario_mgr.validate_uuid(scenario_id) is False:
29 self.send_invalid_resource_id_error(resp)
30 return
31
32 my_leakages = self.scenario_mgr.get(scenario_id).leakages
33 self.send_json_response(resp, my_leakages)
34 except Exception as ex:
35 warnings.warn(str(ex))
36 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
37
[docs]
38 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None:
39 """
40 Adds a new leakage to a given scenario.
41
42 Parameters
43 ----------
44 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_
45 Request instance.
46 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
47 Response instance.
48 scenario_id : `str`
49 UUID of the scenario.
50 """
51 try:
52 if self.scenario_mgr.validate_uuid(scenario_id) is False:
53 self.send_invalid_resource_id_error(resp)
54 return
55
56 leakage = self.load_json_data_from_request(req)
57 if not isinstance(leakage, Leakage):
58 self.send_json_parsing_error(resp)
59 return
60
61 self.scenario_mgr.get(scenario_id).add_leakage(leakage)
62 except Exception as ex:
63 warnings.warn(str(ex))
64 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
65
66
[docs]
67class ScenarioQualityEventHandler(ScenarioBaseHandler):
68 """
69 Class for handling GET and POST requests concerning quality events.
70 """
[docs]
71 def on_get(self, _, resp: falcon.Response, scenario_id: str) -> None:
72 """
73 Gets all quality events of a given scenario.
74
75 Parameters
76 ----------
77 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
78 Response instance.
79 scenario_id : `str`
80 UUID of the scenario.
81 """
82 try:
83 if self.scenario_mgr.validate_uuid(scenario_id) is False:
84 self.send_invalid_resource_id_error(resp)
85 return
86
87 my_system_events = self.scenario_mgr.get(scenario_id).system_events
88 my_quality_events = list(filter(lambda event: isinstance(event, SpeciesInjectionEvent),
89 my_system_events))
90 self.send_json_response(resp, my_quality_events)
91 except Exception as ex:
92 warnings.warn(str(ex))
93 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
94
[docs]
95 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None:
96 """
97 Adds a new quality event to a given scenario.
98
99 Parameters
100 ----------
101 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_
102 Request instance.
103 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
104 Response instance.
105 scenario_id : `str`
106 UUID of the scenario.
107 """
108 try:
109 if self.scenario_mgr.validate_uuid(scenario_id) is False:
110 self.send_invalid_resource_id_error(resp)
111 return
112
113 quality_event = self.load_json_data_from_request(req)
114 if not isinstance(quality_event, SpeciesInjectionEvent):
115 self.send_json_parsing_error(resp)
116 return
117
118 self.scenario_mgr.get(scenario_id).add_system_event(quality_event)
119 except Exception as ex:
120 warnings.warn(str(ex))
121 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
122
123
[docs]
124class ScenarioSensorReadingAttackHandler(ScenarioBaseHandler):
125 """
126 Class for handling GET and POST requests concerning sensor reading attacks.
127 """
[docs]
128 def on_get(self, _, resp: falcon.Response, scenario_id: str) -> None:
129 """
130 Gets all sensor reading attack events of a given scenario.
131
132 Parameters
133 ----------
134 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
135 Response instance.
136 scenario_id : `str`
137 UUID of the scenario.
138 """
139 try:
140 if self.scenario_mgr.validate_uuid(scenario_id) is False:
141 self.send_invalid_resource_id_error(resp)
142 return
143
144 my_sensor_reading_attacks = self.scenario_mgr.get(scenario_id).sensor_reading_attacks
145 self.send_json_response(resp, my_sensor_reading_attacks)
146 except Exception as ex:
147 warnings.warn(str(ex))
148 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
149
[docs]
150 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None:
151 """
152 Adds a new sensor reading attack event to a given scenario.
153
154 Parameters
155 ----------
156 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_
157 Request instance.
158 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
159 Response instance.
160 scenario_id : `str`
161 UUID of the scenario.
162 """
163 try:
164 if self.scenario_mgr.validate_uuid(scenario_id) is False:
165 self.send_invalid_resource_id_error(resp)
166 return
167
168 sensor_reading_attack = self.load_json_data_from_request(req)
169 if not isinstance(sensor_reading_attack, SensorReadingAttack):
170 self.send_json_parsing_error(resp)
171 return
172
173 self.scenario_mgr.get(scenario_id).add_sensor_reading_attack(sensor_reading_attack)
174 except Exception as ex:
175 warnings.warn(str(ex))
176 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
177
178
[docs]
179class ScenarioSensorFaultHandler(ScenarioBaseHandler):
180 """
181 Class for handling GET and POST requests concerning sensor faults.
182 """
[docs]
183 def on_get(self, _, resp: falcon.Response, scenario_id: str) -> None:
184 """
185 Gets all sensor faults of a given scenario.
186
187 Parameters
188 ----------
189 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
190 Response instance.
191 scenario_id : `str`
192 UUID of the scenario.
193 """
194 try:
195 if self.scenario_mgr.validate_uuid(scenario_id) is False:
196 self.send_invalid_resource_id_error(resp)
197 return
198
199 my_sensor_faults = self.scenario_mgr.get(scenario_id).sensor_faults
200 self.send_json_response(resp, my_sensor_faults)
201 except Exception as ex:
202 warnings.warn(str(ex))
203 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR
204
[docs]
205 def on_post(self, req: falcon.Request, resp: falcon.Response, scenario_id: str) -> None:
206 """
207 Adds a new sensor fault to a given scenario.
208
209 Parameters
210 ----------
211 req : `falcon.Request <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#request>`_
212 Request instance.
213 resp : `falcon.Response <https://falcon.readthedocs.io/en/stable/api/request_and_response_asgi.html#response>`_
214 Response instance.
215 scenario_id : `str`
216 UUID of the scenario.
217 """
218 try:
219 if self.scenario_mgr.validate_uuid(scenario_id) is False:
220 self.send_invalid_resource_id_error(resp)
221 return
222
223 sensor_fault = self.load_json_data_from_request(req)
224 if not isinstance(sensor_fault, SensorFault):
225 self.send_json_parsing_error(resp)
226 return
227
228 self.scenario_mgr.get(scenario_id).add_sensor_fault(sensor_fault)
229 except Exception as ex:
230 warnings.warn(str(ex))
231 resp.status = falcon.HTTP_INTERNAL_SERVER_ERROR