Fixed a crash when checking permissions and the ACP has a acod.ty a…

…ttribute set. See #112

CHANGELOG.md

@@ -8,6 +8,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 
+
+## [0.11.3] - 2023-05-17
+
+### Fixed
+- [CSE] Fixed a crash when checking permissions and the ACP has a `acod.ty` attribute set. See #112. Thanks to [samuelbles07](https://github.com/samuelbles07).
+
+
 ## [0.11.2] - 2023-01-23
 
 ### Fixed

README.md

@@ -3,7 +3,7 @@
 # ACME oneM2M CSE
 An open source CSE Middleware for Education.
 
-Version 0.11.2
+Version 0.11.3
 
 [![oneM2M](https://img.shields.io/badge/oneM2M-f00)](https://www.onem2m.org) [![Python](https://img.shields.io/badge/Python-3.8-blue)](https://www.python.org) [![Maintenance](https://img.shields.io/badge/Maintained-Yes-green.svg)](https://github.com/ankraft/ACME-oneM2M-CSE/graphs/commit-activity) [![License](https://img.shields.io/badge/License-BSD%203--Clause-green)](LICENSE) [![MyPy](https://img.shields.io/badge/MyPy-covered-green)](LICENSE)  
 [![Twitter](https://img.shields.io/twitter/url/https/twitter.com/acmeCSE.svg?style=social&label=%40acmeCSE)](https://twitter.com/acmeCSE)

acme/etc/Constants.py

@@ -10,7 +10,7 @@
 class Constants(object):
 	""" Various CSE and oneM2M constants """
 
-	version	= '0.11.2'
+	version	= '0.11.3'
 	"""	ACME's release version """
 
 	textLogo = '[dim][[/dim][red][i]ACME[/i][/red][dim]][/dim]'

acme/resources/ACP.py

@@ -180,9 +180,9 @@ def checkPermission(self, originator:str, requestedPermission:Permission, ty:Res
 						if ty is None or ty not in eachAcod.get('chty'):	# ty is an int
 							continue										# for CREATE: type not in chty
 					else:
-						if ty not in eachAcod.get('ty'):
+						if ty is not None and ty != eachAcod.get('ty'):
 							continue								# any other Permission type: ty not in chty
-					break # found one
+					break # found one, so apply the next checks further down
 				else:
 					continue	# NOT found, so continue the overall search
 

tests/config.py

@@ -7,7 +7,7 @@
 #	Configurations for unit tests
 #
 
-BINDING						= 'mqtt'	# possible values: http, https, mqtt
+BINDING						= 'http'	# possible values: http, https, mqtt
 
 if BINDING == 'mqtt':
 	PROTOCOL				= 'mqtt'	

tests/testACP.py

@@ -665,6 +665,83 @@ def test_retrieveACPwithoutRETRIEVEAccessFail(self) -> None:
 
 
 
+	@unittest.skipIf(noCSE, 'No CSEBase')
+	def test_createACPWithWrongTyFail(self) -> None:
+		"""	Create <ACP> AE originator and wrong acod.ty -> Fail"""
+		dct = 	{ "m2m:acp": {
+					"rn": acpRN,
+					"pv": {
+						"acr": [ {
+							"acor": [ TestACP.originator ],
+							"acop": Permission.CREATE,
+							"acod": [ {
+								"ty": [ T.CIN ],	# acr is only for CIN. Wrong because not a list
+								"chty": [ T.CNT ]	# Allow only a CNT to be created
+							} ]
+						}]
+					},
+					"pvs": { 
+						"acr": [ {
+							"acor": [ TestACP.originator ],
+							"acop": Permission.ALL
+						} ]
+					},
+				}}
+		r, rsc = CREATE(cseURL, TestACP.originator, T.ACP, dct)
+		self.assertEqual(rsc, RC.badRequest, r)
+
+
+	@unittest.skipIf(noCSE, 'No CSEBase')
+	def test_createACPWithTy(self) -> None:
+		"""	Create <ACP> AE originator and acod.ty"""
+		dct = 	{ "m2m:acp": {
+					"rn": acpRN,
+					"pv": {
+						"acr": [ {
+							"acor": [ TestACP.originator ],
+							"acop": Permission.RETRIEVE + Permission.CREATE,
+							"acod": [ {
+								"ty": T.CNT,	# acr is only for CNT
+								"chty": [ T.CNT ]	# Allow only a CNT to be created
+							} ]
+						}]
+					},
+					"pvs": { 
+						"acr": [ {
+							"acor": [ TestACP.originator ],
+							"acop": Permission.ALL
+						} ]
+					},
+				}}
+		r, rsc = CREATE(cseURL, TestACP.originator, T.ACP, dct)
+		self.assertEqual(rsc, RC.created, r)
+		self.assertIsNotNone(findXPath(r, 'm2m:acp/pv/acr/{0}/acod'))
+		self.assertIsNotNone(findXPath(r, 'm2m:acp/pv/acr/{0}/acod/{0}/chty'))
+		self.assertIsInstance(findXPath(r, 'm2m:acp/pv/acr/{0}/acod/{0}/chty'), list)
+		self.assertTrue(T.CNT in findXPath(r, 'm2m:acp/pv/acr/{0}/acod/{0}/chty'))
+		TestACP.acp = r
+
+		# Just delete the CNT
+		DELETE(cntURL, TestACP.originator)
+
+		# Add a container with the acpi set to the ACP
+		dct = { "m2m:cnt": { 
+					"rn": cntRN,
+					"acpi": [ findXPath(TestACP.acp, 'm2m:acp/ri') ]
+			} }
+		r, rsc = CREATE(aeURL, TestACP.originator, T.CNT, dct)
+		self.assertEqual(rsc, RC.created, r)
+
+		# Try to create a CIN -> Fail
+		dct = { "m2m:cin": { "con": "test" } }
+		r, rsc = CREATE(cntURL, TestACP.originator, T.CIN, dct)
+		self.assertEqual(rsc, RC.originatorHasNoPrivilege, r)
+
+		# try to retrieve the CNT -> Fail
+		r, rsc = RETRIEVE(cntURL, TestACP.originator)
+		self.assertEqual(rsc, RC.OK, r)
+
+
 
 
 # TODO reference a non-acp resource in acpi
@@ -728,6 +805,9 @@ def run(testFailFast:bool) -> Tuple[int, int, int, float]:
 
 	addTest(suite, TestACP('test_retrieveACPwithoutRETRIEVEAccessFail'))
 
+	addTest(suite, TestACP('test_createACPWithWrongTyFail'))
+	addTest(suite, TestACP('test_createACPWithTy'))
+
 
 	result = unittest.TextTestRunner(verbosity=testVerbosity, failfast=testFailFast).run(suite)
 	printResult(result)