diff --git a/operations/claims.py b/operations/claims.py
new file mode 100644
index 00000000..00a98d76
--- /dev/null
+++ b/operations/claims.py
@@ -0,0 +1,92 @@
+"""Resources held by jobs and by short requests (spec section 2). A request
+or a job writes its claims in one short transaction, only if no active
+claim conflicts; otherwise it is refused at once (Busy)."""
+
+import os
+import socket
+import threading
+from datetime import timedelta
+
+from django.db import connection, transaction
+from django.db.models import Q
+from django.utils import timezone
+
+from .models import Claim
+
+# gunicorn ends a request after its timeout (gunicorn.conf.py: 600 s); a
+# short operation's claim is certainly orphaned after that.
+TRANSIENT_TTL = timedelta(seconds=600 + 60)
+
+
+class Busy(Exception):
+ """A resource is held. blocker: the job holding it, None for a short operation."""
+
+ def __init__(self, resource, blocker=None):
+ super().__init__(f"{resource} is busy")
+ self.resource = resource
+ self.blocker = blocker
+
+
+def compute_key(compute_id):
+ return f"compute:{compute_id}"
+
+
+def instance_key(uuid):
+ return f"instance:{uuid}"
+
+
+def quota_key(user_id):
+ return f"quota:user:{user_id}"
+
+
+def holder_id():
+ return f"{socket.gethostname()}:{os.getpid()}:{threading.get_ident()}"
+
+
+def serialize(keys):
+ """Inside a transaction: admissions touching one of these keys run one at
+ a time. SQLite: the IMMEDIATE transaction already holds the write lock."""
+ if connection.vendor == "postgresql":
+ with connection.cursor() as cursor:
+ for key in sorted(set(keys)):
+ cursor.execute("SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", [key])
+
+
+def _check(wanted, now):
+ Claim.objects.filter(operation__isnull=True, expires_at__lte=now, resource__in=list(wanted)).delete()
+ active = (
+ Claim.objects.filter(resource__in=list(wanted))
+ .filter(Q(operation__isnull=False) | Q(expires_at__gt=now))
+ .select_related("operation")
+ .order_by("resource")
+ )
+ for claim in active:
+ if wanted[claim.resource] == Claim.EXCLUSIVE or claim.mode == Claim.EXCLUSIVE:
+ raise Busy(claim.resource, claim.operation)
+
+
+def admit(wanted, *, holder, ttl=None):
+ """Transient claims for a short operation; Busy if anything conflicts."""
+ with transaction.atomic():
+ serialize(wanted)
+ now = timezone.now()
+ _check(wanted, now)
+ rows = [Claim(resource=k, mode=m, holder=holder, expires_at=now + (ttl or TRANSIENT_TTL))
+ for k, m in sorted(wanted.items())]
+ return [c.pk for c in Claim.objects.bulk_create(rows)]
+
+
+def admit_job(op, wanted, existing, extra_locks=()):
+ """Save the unsaved job op with its claims. existing() runs first, under
+ the same locks (extra_locks: e.g. the submission key), and its job is
+ returned unchanged when it finds one."""
+ with transaction.atomic():
+ serialize([*wanted, *extra_locks])
+ now = timezone.now()
+ found = existing()
+ if found is not None:
+ return found
+ _check(wanted, now)
+ op.save()
+ Claim.objects.bulk_create([Claim(operation=op, resource=k, mode=m) for k, m in sorted(wanted.items())])
+ return op
diff --git a/operations/tests/test_claims.py b/operations/tests/test_claims.py
new file mode 100644
index 00000000..ae68a8b9
--- /dev/null
+++ b/operations/tests/test_claims.py
@@ -0,0 +1,67 @@
+"""Admission writes claims only when nothing active conflicts, all or
+nothing; exclusive conflicts with any claim, shared only with exclusive."""
+
+from datetime import timedelta
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from django.utils import timezone
+
+from operations import claims
+from operations.models import Claim, Operation
+
+S, X = Claim.SHARED, Claim.EXCLUSIVE
+
+
+def job(user, key="k"):
+ return Operation(kind="test.noop", user=user, username=user.username, submission_key=key,
+ resource="instance:u1", resource_label="c/vm")
+
+
+class AdmitTestCase(TestCase):
+ def setUp(self):
+ self.user = get_user_model().objects.create_user("claims_user", "c@example.com", "pw")
+
+ def test_shared_claims_coexist_and_exclusive_ones_do_not(self):
+ claims.admit({"compute:1": S}, holder="a")
+ claims.admit({"compute:1": S}, holder="b")
+ with self.assertRaises(claims.Busy) as busy:
+ claims.admit({"compute:1": X}, holder="c")
+ self.assertEqual((busy.exception.resource, busy.exception.blocker), ("compute:1", None))
+
+ def test_a_refusal_writes_nothing(self):
+ claims.admit({"instance:u1": X}, holder="a")
+ with self.assertRaises(claims.Busy):
+ claims.admit({"compute:1": S, "instance:u1": X}, holder="b")
+ self.assertFalse(Claim.objects.filter(resource="compute:1").exists())
+
+ def test_a_job_and_its_claims_are_written_together(self):
+ op = claims.admit_job(job(self.user), {"compute:1": S, "instance:u1": X}, existing=lambda: None)
+ self.assertEqual(Operation.objects.get().pk, op.pk)
+ self.assertEqual(set(Claim.objects.values_list("resource", "operation")),
+ {("compute:1", op.pk), ("instance:u1", op.pk)})
+ with self.assertRaises(claims.Busy) as busy:
+ claims.admit({"instance:u1": S}, holder="b")
+ self.assertEqual(busy.exception.blocker.pk, op.pk)
+
+ def test_an_existing_job_is_returned_before_conflicts_are_checked(self):
+ first = claims.admit_job(job(self.user), {"instance:u1": X}, existing=lambda: None)
+ again = claims.admit_job(job(self.user, key="k2"), {"instance:u1": X}, existing=lambda: first)
+ self.assertEqual(again.pk, first.pk)
+ self.assertEqual(Operation.objects.count(), 1)
+
+ def test_a_refused_job_is_not_saved(self):
+ claims.admit({"instance:u1": X}, holder="a")
+ with self.assertRaises(claims.Busy):
+ claims.admit_job(job(self.user), {"instance:u1": X}, existing=lambda: None)
+ self.assertFalse(Operation.objects.exists())
+
+ def test_an_expired_transient_claim_does_not_block(self):
+ Claim.objects.create(resource="instance:u1", mode=X, holder="dead:1:1",
+ expires_at=timezone.now() - timedelta(seconds=1))
+ claims.admit({"instance:u1": X}, holder="b")
+ self.assertEqual(list(Claim.objects.values_list("holder", flat=True)), ["b"])
+
+ def test_keys_are_canonical(self):
+ self.assertEqual(claims.compute_key(3), "compute:3")
+ self.assertEqual(claims.instance_key("u-1"), "instance:u-1")
diff --git a/operations/tests/test_redact.py b/operations/tests/test_redact.py
new file mode 100644
index 00000000..dbb71dc0
--- /dev/null
+++ b/operations/tests/test_redact.py
@@ -0,0 +1,36 @@
+from django.test import SimpleTestCase
+
+from operations import utils
+
+
+class RedactTestCase(SimpleTestCase):
+ def test_credentials_in_text_are_hidden(self):
+ text = ("Cannot connect to qemu+tcp://admin:s3cret@192.0.2.5/system: "
+ " password=hunter2 auth_password: abc console_pass=cp1 "
+ 'password="quoted1" token: \'quoted2\' pp')
+ safe = utils.safe_text(text)
+ for secret in ("s3cret", "vncpw", "hunter2", "abc", "quoted1", "quoted2", "pp<", "cp1"):
+ self.assertNotIn(secret, safe)
+ self.assertIn("192.0.2.5", safe)
+
+ def test_structured_and_multiline_secrets_are_hidden(self):
+ safe = utils.safe_text('{"password": "hunter2", \'token\': \'t0k\'} password="line1\nline2"')
+ for secret in ("hunter2", "t0k", "line1", "line2"):
+ self.assertNotIn(secret, safe)
+
+ def test_escaped_quotes_and_nested_markup_are_hidden(self):
+ safe = utils.safe_text('{"password": "abc\\"def"} y]]>')
+ for secret in ("abc", "def", "x<", "y]]"):
+ self.assertNotIn(secret, safe)
+
+ def test_unterminated_and_pathological_quotes_are_handled(self):
+ self.assertNotIn("hunter2", utils.safe_text('failed: password="hunter2'))
+ utils.safe_text('password="' + "\\" * 5000)
+
+ def test_secret_looking_params_are_refused(self):
+ for key in ("password", "root_pass", "console_passwd", "secret", "private_key"):
+ with self.subTest(key), self.assertRaises(ValueError):
+ utils.check_params({"name": "x", key: "v"})
+ with self.assertRaises(ValueError):
+ utils.check_params({"config": [{"password": "v"}]})
+ utils.check_params({"instance": 1, "name": "snap"})
diff --git a/operations/utils.py b/operations/utils.py
new file mode 100644
index 00000000..ef67fb23
--- /dev/null
+++ b/operations/utils.py
@@ -0,0 +1,32 @@
+"""Secrets never reach a job record, a response or a log line."""
+
+import re
+
+_NAMES = r"(?:pass|secret|token|key)\w*"
+_PATTERNS = [
+ (re.compile(r"(://[^:/@\s]+:)[^@\s]+@"), r"\1***@"),
+ # quoted values, also XML attributes: passwd='x', password="x"
+ (re.compile(rf"""({_NAMES}['"]?\s*[:=]\s*)(['"])(?:\\.|(?!\2)[^\\])*\\?(?:\2|\Z)""", re.I | re.S), r"\1\2***\2"),
+ (re.compile(rf"""({_NAMES}['"]?\s*[:=]\s*)[^\s'",;<>]+""", re.I), r"\1***"),
+ # value-like elements
+ (re.compile(r"(<(secret|passphrase)\b[^>]*>).*?(\2>)", re.I | re.S), r"\1***\3"),
+]
+_SECRET_KEY = re.compile(r"pass|secret|token|private|key$", re.I)
+
+
+def safe_text(text):
+ text = str(text)
+ for pattern, replacement in _PATTERNS:
+ text = pattern.sub(replacement, text)
+ return text
+
+
+def check_params(params):
+ if isinstance(params, dict):
+ for key, value in params.items():
+ if _SECRET_KEY.search(str(key)):
+ raise ValueError(f"A job cannot store the field {key}")
+ check_params(value)
+ elif isinstance(params, (list, tuple)):
+ for value in params:
+ check_params(value)