Put the Workspace Check Inside the Database Write
A record can belong to workspace Alpha when you check it and workspace Beta when you update it. If the UPDATE filters only by record ID, the earlier check no longer protects the write. In the comments on the previous MC
A record can belong to workspace Alpha when you check it and workspace Beta when you update it. If the UPDATE filters only by record ID, the earlier check no longer protects the write.
In the comments on the previous MCP permissions example, arun pointed out this database boundary. That example used an in-memory map. Moving the handler to a database needs one more decision: which rows is the mutation itself allowed to touch?
Here is a small SQLite example that makes the gap reproducible, then closes it with a workspace-scoped write.
A successful check can become stale
Imagine this sequence:
| Step | Connection A | Connection B |
|---|---|---|
| 1 | Reads alpha-job; its workspace is alpha
|
|
| 2 | Transfers alpha-job to beta and commits |
|
| 3 | Updates alpha-job by ID |
This is the unsafe last statement:
UPDATE automations
SET interval_minutes = ?
WHERE id = ?;
The ID still matches after the transfer. The statement contains no condition that would reject the row's new workspace.
The local test below deliberately runs that sequence. Its vulnerable control confirms that the row now belongs to Beta and still receives Alpha's requested interval. A passing control means the unwanted behavior was reproduced.
Put the workspace in the mutation
The guarded statement is:
UPDATE automations
SET interval_minutes = ?
WHERE id = ? AND workspace_id = ?
RETURNING id, workspace_id, interval_minutes, running;
The third parameter comes from the authenticated principal supplied by trusted server code. It is not another field the model may choose in its tool arguments.
SQLite updates only rows whose WHERE condition is true. If the row has moved to Beta before this statement executes, the Alpha predicate matches nothing.
That is a normal zero-row database result, so the application must handle it deliberately. It must not return a success message just because the SQL statement completed without an exception.
Return the row that was actually updated
SQLite's RETURNING clause produces a result row for each directly affected row. The function below consumes and closes the cursor before returning. With these autocommit connections, it does not leave an explicit transaction for the caller to commit later.
No returned row becomes Unavailable("Automation is not available."). The same message covers an absent ID and an ID in another workspace; the message does not reveal which case occurred.
The function also checks the write scope and validates the interval. The range of 5–1440 minutes is a constraint chosen for this demonstration. It is not a claim about a product's settings.
The SQL changes only interval_minutes. It does not start the automation. The tests inspect the saved row, the untouched row in the other workspace, and the running value.
Run the example
Save the following as guarded_workspace_write.py and run:
python guarded_workspace_write.py
It uses Python's standard library. You need Python 3.9 or later and SQLite 3.35 or later for RETURNING.
Both connections use isolation_level=None. The test finishes A's initial SELECT, then lets B commit the transfer, then lets A attempt the write. There is no long-lived read transaction holding the old result open.
This is a controlled interleaving between two real database connections. It makes the stale-check failure repeatable without depending on thread timing. SQLite serializes writers; this example does not assume that two writes execute simultaneously.
"""A local SQLite teaching example; it is not a SMTasker implementation."""
from dataclasses import dataclass
from pathlib import Path
import sqlite3
import sys
import tempfile
import unittest
@dataclass(frozen=True)
class Principal:
workspace_id: str
scopes: frozenset[str]
class Unavailable(Exception):
pass
def set_interval(db, principal, automation_id, minutes):
# principal is supplied by trusted authentication code, not tool arguments.
if "automation:write" not in principal.scopes:
raise PermissionError("Write access required.")
if type(minutes) is not int or not 5 <= minutes <= 1440:
raise ValueError("interval_minutes must be an integer from 5 to 1440.")
cursor = db.execute(
"""UPDATE automations
SET interval_minutes = ?
WHERE id = ? AND workspace_id = ?
RETURNING id, workspace_id, interval_minutes, running""",
(minutes, automation_id, principal.workspace_id),
)
try:
rows = cursor.fetchall()
finally:
cursor.close()
if not rows:
# Same response for an absent row and an inaccessible row.
raise Unavailable("Automation is not available.")
return dict(rows[0])
class WorkspaceWriteTests(unittest.TestCase):
def setUp(self):
fixture_root = Path(__file__).resolve().parent / "test-db"
fixture_root.mkdir(exist_ok=True)
self.tmp = tempfile.TemporaryDirectory(prefix="workspace-", dir=fixture_root)
# Cleanup is confined to the generated fixture under this project.
assert Path(self.tmp.name).resolve().is_relative_to(fixture_root.resolve())
database = str(Path(self.tmp.name) / "example.sqlite")
self.a = sqlite3.connect(database, isolation_level=None)
self.b = sqlite3.connect(database, isolation_level=None)
self.a.row_factory = sqlite3.Row
self.b.row_factory = sqlite3.Row
self.a.executescript(
"""CREATE TABLE automations (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL,
interval_minutes INTEGER NOT NULL
CHECK(interval_minutes BETWEEN 5 AND 1440),
running INTEGER NOT NULL CHECK(running IN (0, 1))
);
INSERT INTO automations VALUES ('alpha-job', 'alpha', 60, 0);
INSERT INTO automations VALUES ('beta-job', 'beta', 90, 0);"""
)
self.editor = Principal("alpha", frozenset({"automation:write"}))
def tearDown(self):
self.a.close()
self.b.close()
self.tmp.cleanup()
def row(self, automation_id):
cursor = self.b.execute(
"SELECT * FROM automations WHERE id = ?", (automation_id,)
)
try:
found = cursor.fetchone()
return None if found is None else dict(found)
finally:
cursor.close()
def transfer_after_read(self):
cursor = self.a.execute(
"SELECT workspace_id FROM automations WHERE id = ?", ("alpha-job",)
)
try:
observed_workspace = cursor.fetchone()["workspace_id"]
finally:
cursor.close()
self.assertEqual(observed_workspace, "alpha")
# Another committed operation changes ownership before A writes.
self.b.execute(
"UPDATE automations SET workspace_id = 'beta' WHERE id = 'alpha-job'"
).close()
def test_authorized_update_preserves_other_fields_and_rows(self):
beta_before = self.row("beta-job")
result = set_interval(self.a, self.editor, "alpha-job", 30)
self.assertEqual(result, {
"id": "alpha-job", "workspace_id": "alpha",
"interval_minutes": 30, "running": 0,
})
self.assertEqual(self.row("alpha-job"), result)
self.assertEqual(self.row("beta-job"), beta_before)
def test_read_only_principal_cannot_write(self):
before = self.row("alpha-job")
reader = Principal("alpha", frozenset({"automation:read"}))
with self.assertRaises(PermissionError):
set_interval(self.a, reader, "alpha-job", 30)
self.assertEqual(self.row("alpha-job"), before)
def test_unsafe_id_only_write_reproduces_the_gap(self):
self.transfer_after_read()
self.a.execute(
"UPDATE automations SET interval_minutes = ? WHERE id = ?",
(30, "alpha-job"),
).close()
row = self.row("alpha-job")
self.assertEqual((row["workspace_id"], row["interval_minutes"]), ("beta", 30))
def test_guarded_write_refuses_a_transferred_row(self):
self.transfer_after_read()
before = self.row("alpha-job")
with self.assertRaises(Unavailable):
set_interval(self.a, self.editor, "alpha-job", 30)
self.assertEqual(self.row("alpha-job"), before)
self.assertEqual(before["interval_minutes"], 60)
def test_other_workspace_is_unavailable(self):
before = self.row("beta-job")
with self.assertRaisesRegex(Unavailable, "^Automation is not available\\.$"):
set_interval(self.a, self.editor, "beta-job", 30)
self.assertEqual(self.row("beta-job"), before)
def test_missing_row_uses_the_same_unavailable_error(self):
with self.assertRaisesRegex(Unavailable, "^Automation is not available\\.$"):
set_interval(self.a, self.editor, "missing-job", 30)
def test_invalid_intervals_are_rejected_without_writes(self):
before = self.row("alpha-job")
for value in (True, "30", 0, 1441):
with self.subTest(value=value):
with self.assertRaises(ValueError):
set_interval(self.a, self.editor, "alpha-job", value)
self.assertEqual(self.row("alpha-job"), before)
def test_bound_id_is_not_sql(self):
before = (self.row("alpha-job"), self.row("beta-job"))
with self.assertRaises(Unavailable):
set_interval(self.a, self.editor, "alpha-job' OR 1=1 --", 30)
self.assertEqual((self.row("alpha-job"), self.row("beta-job")), before)
if __name__ == "__main__":
if sqlite3.sqlite_version_info < (3, 35, 0):
raise SystemExit("This example needs SQLite 3.35 or later for RETURNING.")
print(f"Python {sys.version.split()[0]}; SQLite {sqlite3.sqlite_version}", flush=True)
unittest.main(verbosity=2)
What the tests establish
The eight tests cover:
- An authorized update changes the requested interval while preserving the other row and the stopped state.
- A principal with read access cannot write.
- An ID-only update reproduces the stale-ownership failure.
- The workspace-scoped update refuses the same transferred row without changing its interval.
- Another workspace's row is unavailable to the caller.
- A missing row produces the same application-level rejection message.
- Invalid interval values leave the stored row unchanged.
- An ID containing SQL-like text remains a bound value and updates nothing.
The parameter-binding test and the workspace test protect different boundaries. Binding a value safely does not decide whether the caller owns the row.
Keep the claim narrow
This guard addresses the ownership condition on this write path. The principal still needs trustworthy authentication, and the service still needs scope checks, allowed-field checks and domain validation.
It does not implement membership revocation or another approval system. Two authorized editors in the same workspace can also need separate lost-update protection, such as an expected-version condition. Those requirements belong in the data-layer design too.
Review the ownership policy on every write path: MCP, HTTP endpoints, background jobs and admin actions. Deliberately privileged operations need an explicit policy too; they should not inherit an unchecked ID-only write accidentally.
The practical change is small: keep the early check if it helps the flow, and make the mutation enforce the workspace condition on the row it changes.
Does your data layer have another write path that needs the same workspace predicate? Leave a small example in the comments.
Originally published by Dev.to Security. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.