Broken ID usage for BACNetID.

[terjekv]

The BACnetID model has a first_unused_id() static method:

mreg/mreg/models/host.py

Lines 78 to 97 in bbfcf18

	class BACnetID(models.Model):
	id = models.IntegerField(primary_key=True, validators=[validate_BACnetID])
	host = models.OneToOneField(Host, on_delete=models.CASCADE, related_name="bacnetid")
	class Meta:
	db_table = "bacnetid"
	@property
	def hostname(self):
	return self.host.name
	@staticmethod
	def first_unused_id() -> int:
	j = 0
	for i in BACnetID.objects.values_list("id", flat=True).order_by("id"):
	if i == j:
	j += 1
	else:
	return j
	return j

first_unused_id()does not look like anything that should ever exist... Now, it is used in tests:

mreg/mreg/api/v1/tests/tests_bacnet.py

Lines 17 to 22 in bbfcf18

	def setUp(self):
	"""Define the test client and other test variables."""
	super().setUp()
	self.host_one = Host(name='host1.example.org', contact='[email protected]')
	self.host_one.save()
	self.id_one = BACnetID.objects.create(id=BACnetID.first_unused_id(), host=self.host_one)

and

mreg/mreg/api/v1/tests/tests_bacnet.py

Lines 30 to 32 in bbfcf18

	def test_list_200_ok(self):
	"""List all should return 200"""
	self.id_two = BACnetID.objects.create(id=BACnetID.first_unused_id(), host=self.host_two)

But, much worse is that the code is used during creation of BACnetID entries:

mreg/mreg/api/v1/views_bacnet.py

Lines 23 to 62 in bbfcf18

	def post(self, request, *args, **kwargs):
	# request.data is immutable
	data = request.data.copy()
	# if no ID value was supplied, pick the next available ID value
	if "id" not in data:
	data["id"] = BACnetID.first_unused_id()
	else:
	# if an ID value was supplied, and it is already in use, return 409 conflict
	# instead of the default 400 bad request
	if BACnetID.objects.filter(id=data["id"]).exists():
	return Response(status=status.HTTP_409_CONFLICT)
	try:
	# allow clients to supply a hostname instead of a host id
	host = None
	if "hostname" in data:
	host = Host.objects.get(name=data["hostname"])
	data["host"] = host.id
	elif "host" in data:
	host = Host.objects.get(id=data["host"])
	# if a host was supplied and that host already has a BACnet ID, return 409 conflict
	# instead of the default 400 bad request
	if host and hasattr(host, "bacnetid"):
	content = {"ERROR": "The host already has a BACnet ID."}
	return Response(content, status=status.HTTP_409_CONFLICT)
	except Host.DoesNotExist:
	content = {"ERROR": "The host does not exist."}
	return Response(content, status=status.HTTP_400_BAD_REQUEST)
	# validate the data
	obj = BACnetID()
	ser = serializers.BACnetIDSerializer(obj, data=data)
	if ser.is_valid(raise_exception=True):
	# create a new object
	self.perform_create(ser)
	location = request.path + str(obj.id)
	return Response(
	status=status.HTTP_201_CREATED, headers={"Location": location}
	)

This is a fairly classical example of a race condition and should be remedied.