from defence360agent.model.simplification import remove_old_and_truncate from defence360agent.rpc_tools import ValidationError, lookup from defence360agent.rpc_tools.utils import run_in_executor_decorator import defence360agent.subsys.panels.hosting_panel as hp from defence360agent.utils import Scope from im360.contracts import config from im360.model.firewall import Purpose from im360.model.incident import Incident class IncidentsEndpoints(lookup.RootEndpoints): SCOPE = Scope.IM360 @lookup.bind("get") async def get_incident_list(self, *args, **kwargs): if ("since" not in kwargs) and ("period" not in kwargs): raise ValidationError("Either since or period should be specified") period = kwargs.pop("period", None) if period is not None: kwargs["since"] = period.since kwargs["to"] = period.to severity = kwargs.pop("severity", None) if kwargs.get("by_purpose"): # backward compatibility kwargs["by_list"] = [ Purpose.listname(item) for item in kwargs.pop("by_purpose") ] by_panel_user = kwargs.pop("by_panel_user", None) user_domains = None if by_panel_user is not None: user_domains = ( await hp.HostingPanel().get_domains_per_user() ).get(by_panel_user, []) return Incident.get_sorted_incident_list( *args, severity=severity or config.IncidentLogging.MIN_LOG_LEVEL, by_domains=user_domains, **kwargs ) @lookup.bind("clean") @run_in_executor_decorator def clean_incidents(self, days=None, limit=None): return remove_old_and_truncate( Incident, days or config.IncidentLogging.NUM_DAYS, limit or config.IncidentLogging.LIMIT, )