openc2lib.actuators.iptables_actuator

Skeleton Actuator for SLPF profile

This module provides an example to create an Actuator for the SLPF profile. It only answers to the request for available features.

View Source

 1""" Skeleton `Actuator` for SLPF profile
 2
 3	This module provides an example to create an `Actuator` for the SLPF profile.
 4	It only answers to the request for available features.
 5"""
 6from openc2lib import ArrayOf,ActionTargets, TargetEnum, Nsid, Version,Actions, Command, Response, StatusCode, StatusCodeDescription
 7
 8import openc2lib.profiles.slpf as slpf 
 9
10
11OPENC2VERS=Version(1,0)
12""" Supported OpenC2 Version """
13
14# An implementation of the slpf profile. 
15class IptablesActuator:
16	""" Dumb SLPF implementation
17
18		This class provides a skeleton for implementing an `Actuator` according to the openc2lib approach.
19	"""
20	profile = slpf
21
22
23	def run(self, cmd):
24		""" Process `Command`
25
26			The `run` method executes an OpenC2 `Command` and returns a `Response`.
27		"""
28		if not slpf.validate_command(cmd):
29			return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Invalid Action/Target pair')
30
31		if not slpf.validate_args(cmd):
32			return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Option not supported')
33
34		match cmd.action:
35			case Actions.query:
36				r = self.query(cmd)
37			case _:
38				r = self.__notimplemented(cmd)	
39
40		return r
41
42	def query(self, cmd):
43		""" Query action
44
45			This method implements the `query` action.
46			:param cmd: The `Command` including `Target` and optional `Args`.
47			:return: A `Response` including the result of the query and appropriate status code and messages.
48		"""
49		at = ActionTargets()
50		pf = ArrayOf(Nsid)()
51		pf.append(Nsid('slpf'))
52		res = slpf.Results(versions=ArrayOf(Version)([OPENC2VERS]), 
53				profiles=pf, pairs=slpf.AllowedCommandTarget)
54		r = Response(status=StatusCode.OK, status_text=StatusCodeDescription[StatusCode.OK], results=res)
55
56		return r
57
58	def allow(self, cmd):
59		pass
60
61	def deny(self, cmd):
62		pass
63
64	def update(self, cmd):
65		pass
66
67	def __notimplemented(self, cmd):
68		""" Default response
69
70			Default response returned in case an `Action` is not implemented.
71			The `cmd` argument is only present for uniformity with the other handlers.
72			:param cmd: The `Command` that triggered the error.
73			:return: A `Response` with the appropriate error code.
74
75		"""
76		return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Command not implemented')

OPENC2VERS = '1.0'

Supported OpenC2 Version

class IptablesActuator: View Source

16class IptablesActuator:
17	""" Dumb SLPF implementation
18
19		This class provides a skeleton for implementing an `Actuator` according to the openc2lib approach.
20	"""
21	profile = slpf
22
23
24	def run(self, cmd):
25		""" Process `Command`
26
27			The `run` method executes an OpenC2 `Command` and returns a `Response`.
28		"""
29		if not slpf.validate_command(cmd):
30			return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Invalid Action/Target pair')
31
32		if not slpf.validate_args(cmd):
33			return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Option not supported')
34
35		match cmd.action:
36			case Actions.query:
37				r = self.query(cmd)
38			case _:
39				r = self.__notimplemented(cmd)	
40
41		return r
42
43	def query(self, cmd):
44		""" Query action
45
46			This method implements the `query` action.
47			:param cmd: The `Command` including `Target` and optional `Args`.
48			:return: A `Response` including the result of the query and appropriate status code and messages.
49		"""
50		at = ActionTargets()
51		pf = ArrayOf(Nsid)()
52		pf.append(Nsid('slpf'))
53		res = slpf.Results(versions=ArrayOf(Version)([OPENC2VERS]), 
54				profiles=pf, pairs=slpf.AllowedCommandTarget)
55		r = Response(status=StatusCode.OK, status_text=StatusCodeDescription[StatusCode.OK], results=res)
56
57		return r
58
59	def allow(self, cmd):
60		pass
61
62	def deny(self, cmd):
63		pass
64
65	def update(self, cmd):
66		pass
67
68	def __notimplemented(self, cmd):
69		""" Default response
70
71			Default response returned in case an `Action` is not implemented.
72			The `cmd` argument is only present for uniformity with the other handlers.
73			:param cmd: The `Command` that triggered the error.
74			:return: A `Response` with the appropriate error code.
75
76		"""
77		return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Command not implemented')

Dumb SLPF implementation

This class provides a skeleton for implementing an Actuator according to the openc2lib approach.

def run(self, cmd): View Source

24	def run(self, cmd):
25		""" Process `Command`
26
27			The `run` method executes an OpenC2 `Command` and returns a `Response`.
28		"""
29		if not slpf.validate_command(cmd):
30			return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Invalid Action/Target pair')
31
32		if not slpf.validate_args(cmd):
33			return Response(status=StatusCode.NOTIMPLEMENTED, status_text='Option not supported')
34
35		match cmd.action:
36			case Actions.query:
37				r = self.query(cmd)
38			case _:
39				r = self.__notimplemented(cmd)	
40
41		return r

Process Command

The run method executes an OpenC2 Command and returns a Response.

def query(self, cmd): View Source

43	def query(self, cmd):
44		""" Query action
45
46			This method implements the `query` action.
47			:param cmd: The `Command` including `Target` and optional `Args`.
48			:return: A `Response` including the result of the query and appropriate status code and messages.
49		"""
50		at = ActionTargets()
51		pf = ArrayOf(Nsid)()
52		pf.append(Nsid('slpf'))
53		res = slpf.Results(versions=ArrayOf(Version)([OPENC2VERS]), 
54				profiles=pf, pairs=slpf.AllowedCommandTarget)
55		r = Response(status=StatusCode.OK, status_text=StatusCodeDescription[StatusCode.OK], results=res)
56
57		return r

Query action

This method implements the query action.

Parameters

cmd: The Command including Target and optional Args.

Returns

A Response including the result of the query and appropriate status code and messages.

def allow(self, cmd): View Source

59	def allow(self, cmd):
60		pass

def deny(self, cmd): View Source

62	def deny(self, cmd):
63		pass

def update(self, cmd): View Source

65	def update(self, cmd):
66		pass