"""Endpoints for configuration management commands. - manage ModSecurity directives, rulesets. """ import logging from collections import namedtuple from defence360agent.rpc_tools import lookup from defence360agent.rpc_tools.validate import ValidationError from defence360agent.utils import Scope from im360.contracts import config from im360.subsys.panels import hosting_panel __all__ = [] logger = logging.getLogger(__name__) ISSUE_ID_MODSEC_DIRECTIVE_WRONG_VALUE = "1000" MODSEC_DIRECTIVES = dict( (attr, getattr(config.ModSecurityDirectives, attr)) for attr in dir(config.ModSecurityDirectives) if attr.startswith("Sec") ) Issue = namedtuple("Issue", "id title fix url ignored") issues_modsec_directives = [ ( directive, expected, Issue( ISSUE_ID_MODSEC_DIRECTIVE_WRONG_VALUE, "Wrong value for {directive} ModSecurity directive." " Expected: {expected!r} Got: {{got!r}}".format(**vars()), "Run `imunify360-agent fix modsec directives` command", "https://docs.imunify360.com/issues/" + ISSUE_ID_MODSEC_DIRECTIVE_WRONG_VALUE, ignored=False, ), ) for directive, expected in MODSEC_DIRECTIVES.items() ] class ModSecEndpoints(lookup.RootEndpoints): SCOPE = Scope.IM360 def __init__(self, sink): super().__init__(sink) self.hosting_panel = hosting_panel.HostingPanel() @lookup.bind("check", "modsec", "directives") async def validate_modsec_directives(self): skip = config.StopManaging.MODSEC_DIRECTIVES issues = [] for directive, expected, issue in issues_modsec_directives: got = await self.hosting_panel.modsec_get_directive(directive) if got != expected: logger.debug( "directive %r got %r expected %r", directive, got, expected, ) issue = issue._replace(ignored=skip) issues.append( dict(issue._asdict(), title=issue.title.format(**vars())) ) if not all(issue["ignored"] for issue in issues): # there are issues that not ignored raise ValidationError(issues) # produce warnings return {"items": issues} # no issues or all issues are ignored @lookup.bind("fix", "modsec", "directives") async def reset_modsec_directives(self): if config.StopManaging.MODSEC_DIRECTIVES: return { "items": [ "Skipped resetting ModSecurity directives." "\nTo reenable the command, run:" """\n imunify360-agent config update '{"STOP_MANAGING": {"modsec_directives": false}}'""" # noqa ] } await self.hosting_panel.reset_modsec_directives() return {"items": []} @lookup.bind("check", "modsec", "rulesets") async def validate_modsec_rulesets(self): raise NotImplementedError @lookup.bind("fix", "modsec", "rulesets") async def reset_modsec_rulesets(self): raise NotImplementedError