DEV Community: SMTasker Team

Put the Workspace Check Inside the Database Write

SMTasker Team — Sun, 01 Nov 2026 00:22:37 +0000

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.

Enforcing MCP Permissions Before a Tool Handler Runs

SMTasker Team — Wed, 28 Oct 2026 23:20:24 +0000

A tool description contains readOnlyHint: true. The assistant asks to inspect a stopped automation. The operator expects the configuration to stay unchanged.

That expectation needs enforcement at the service handling the request. A label in a tool definition cannot remove write access from a credential or prevent a handler from changing a record.

The Model Context Protocol, or MCP, lets applications expose tools to AI clients. Its ToolAnnotations specification treats annotations as hints about behavior. They can help a client present a tool or decide when to ask for confirmation. They do not establish what the caller is authorized to do. The MCP maintainers explain this distinction in their discussion of annotations and enforcement.

For a developer exposing an operational tool, the useful question is: where does an unauthorized request stop before the handler touches state?

Separate the tool description from the access decision

Consider a fictional service with two tools:

Tool	Intended behavior	Required permission
`automation.inspect`	Return one automation's configuration	`automation:read`
`automation.set_interval`	Change its execution interval	`automation:write`

Those scope names are invented for this example. They are application policy, not standard MCP scopes or SMTasker API fields.

The service needs to identify the authenticated caller, check the required permission, and confirm that the requested record belongs to the caller's workspace. Checking only the tool name leaves the resource boundary unresolved: an editor for workspace Alpha should not be able to update workspace Beta by changing an ID.

The MCP tools specification requires input validation and access controls on servers. The following example isolates those checks so you can inspect their order.

Check permission and resource ownership before dispatch

This is an in-memory teaching model. It does not implement an MCP transport, token validation, a database or SMTasker's authorization system. The principal represents information that trusted authentication middleware would supply; it must never be accepted from the model's tool-call arguments.

The important boundary is inside execute. It looks up server-owned policy, checks the required scope, validates the argument shape, checks ownership of the stored record, and validates the requested value. Only then can the handler run.

function execute(principal, name, args) {
  const tool = tools.get(name);
  if (!tool) throw new Error('Unknown tool');
  if (!principal || !principal.scopes.has(tool.scope)) {
    throw new Error('Forbidden');
  }
  if (!args || typeof args !== 'object' || Array.isArray(args) ||
      typeof args.automationId !== 'string' ||
      Object.keys(args).some(key => !tool.fields.includes(key))) {
    throw new Error('Invalid input');
  }
  const record = records.get(args.automationId);
  if (!record || record.workspaceId !== principal.workspaceId) {
    throw new Error('Not found or unauthorized');
  }
  tool.validate(args);
  return tool.run(record, args);
}

Notice that this decision does not read annotations.readOnlyHint. The server's policy determines the required permission. An annotation still describes the tool for clients, but changing that annotation cannot turn a reader into an editor.

Ownership also comes from the stored record. A caller cannot supply a different workspaceId to make the authorization check pass. The example rejects that extra argument.

Test the requests that must be rejected

A useful demonstration needs more than a successful update. Each rejected write should leave the stored records unchanged.

The companion example includes these checks:

Request	Expected result
Reader inspects an Alpha record	Returns a copy without changing stored state
Reader requests an interval change	Denied before mutation
Write tool incorrectly advertises a read-only hint	Reader's write remains denied
Reader inspects a Beta record	Denied by the resource boundary
Alpha editor modifies a Beta record	Denied without changing either workspace
Caller supplies a workspace override	Rejected as invalid input
Interval is missing, a string, fractional or outside the accepted range	Rejected without mutation
Alpha editor changes an Alpha interval	Only the interval changes; the job stays stopped
Tool is unknown or the principal is absent	Denied

Save the following as permission-boundary-demo.mjs and run it with Node.js:

node permission-boundary-demo.mjs

import assert from 'node:assert/strict';

// Teaching model only: no network, credentials or real product API.
// In a service, trusted authentication middleware supplies this principal.
const reader = { workspaceId: 'alpha', scopes: new Set(['automation:read']) };
const editor = { workspaceId: 'alpha', scopes: new Set(['automation:read', 'automation:write']) };
const records = new Map([
  ['alpha-job', { workspaceId: 'alpha', intervalMinutes: 60, running: false }],
  ['beta-job', { workspaceId: 'beta', intervalMinutes: 60, running: false }],
]);

// This is server-owned policy and implementation, never tool-call input.
const tools = new Map([
  ['automation.inspect', {
    scope: 'automation:read',
    annotations: { readOnlyHint: true },
    fields: ['automationId'],
    validate() {},
    run: record => ({ ...record }),
  }],
  ['automation.set_interval', {
    scope: 'automation:write',
    annotations: { readOnlyHint: false },
    fields: ['automationId', 'intervalMinutes'],
    validate(args) {
      if (!Number.isInteger(args.intervalMinutes) ||
          args.intervalMinutes < 1 || args.intervalMinutes > 1440) {
        throw new Error('Invalid interval');
      }
    },
    run(record, args) {
      record.intervalMinutes = args.intervalMinutes;
      return { ...record };
    },
  }],
]);

function execute(principal, name, args) {
  const tool = tools.get(name);
  if (!tool) throw new Error('Unknown tool');
  if (!principal || !principal.scopes.has(tool.scope)) {
    throw new Error('Forbidden');
  }
  if (!args || typeof args !== 'object' || Array.isArray(args) ||
      typeof args.automationId !== 'string' ||
      Object.keys(args).some(key => !tool.fields.includes(key))) {
    throw new Error('Invalid input');
  }
  const record = records.get(args.automationId);
  if (!record || record.workspaceId !== principal.workspaceId) {
    throw new Error('Not found or unauthorized');
  }
  tool.validate(args);
  return tool.run(record, args);
}

let passed = 0;
function check(name, action) {
  const before = structuredClone(records);
  try {
    action();
    passed += 1;
    console.log(`PASS ${name}`);
  } finally {
    records.clear();
    for (const [id, value] of before) records.set(id, value);
  }
}

check('read returns a copy and leaves stored state unchanged', () => {
  const before = structuredClone(records);
  const result = execute(reader, 'automation.inspect', { automationId: 'alpha-job' });
  result.intervalMinutes = 5;
  assert.deepEqual(records, before);
});

check('reader cannot write and denial leaves state unchanged', () => {
  const before = structuredClone(records);
  assert.throws(() => execute(reader, 'automation.set_interval', {
    automationId: 'alpha-job', intervalMinutes: 30,
  }), /Forbidden/);
  assert.deepEqual(records, before);
});

check('incorrect read-only annotation cannot grant write access', () => {
  const writer = tools.get('automation.set_interval');
  writer.annotations.readOnlyHint = true;
  const before = structuredClone(records);
  try {
    assert.throws(() => execute(reader, 'automation.set_interval', {
      automationId: 'alpha-job', intervalMinutes: 30,
    }), /Forbidden/);
    assert.deepEqual(records, before);
  } finally {
    writer.annotations.readOnlyHint = false;
  }
});

check('reader cannot inspect another workspace', () => {
  assert.throws(() => execute(reader, 'automation.inspect', {
    automationId: 'beta-job',
  }), /Not found or unauthorized/);
});

check('editor cannot modify another workspace', () => {
  const before = structuredClone(records);
  assert.throws(() => execute(editor, 'automation.set_interval', {
    automationId: 'beta-job', intervalMinutes: 30,
  }), /Not found or unauthorized/);
  assert.deepEqual(records, before);
});

check('caller cannot add a workspace override', () => {
  assert.throws(() => execute(editor, 'automation.set_interval', {
    automationId: 'beta-job', workspaceId: 'alpha', intervalMinutes: 30,
  }), /Invalid input/);
});

check('invalid interval cannot change the record', () => {
  const before = structuredClone(records);
  for (const value of [0, -1, 1.5, 1441, '30', null, undefined]) {
    assert.throws(() => execute(editor, 'automation.set_interval', {
      automationId: 'alpha-job', intervalMinutes: value,
    }), /Invalid interval/);
  }
  assert.deepEqual(records, before);
});

check('authorized write changes only the requested field', () => {
  const other = structuredClone(records.get('beta-job'));
  const result = execute(editor, 'automation.set_interval', {
    automationId: 'alpha-job', intervalMinutes: 30,
  });
  assert.deepEqual(result, { workspaceId: 'alpha', intervalMinutes: 30, running: false });
  assert.deepEqual(records.get('alpha-job'), result);
  assert.deepEqual(records.get('beta-job'), other);
});

check('unknown tool and missing principal fail closed', () => {
  assert.throws(() => execute(editor, 'toString', {}), /Unknown tool/);
  assert.throws(() => execute(null, 'automation.inspect', {
    automationId: 'alpha-job',
  }), /Forbidden/);
});

console.log(`${passed} checks passed. No network requests or external changes.`);

The final line reports nine checks passed. These are checks of this local model. To evaluate a deployed service, exercise the same boundaries through its actual entry point with test credentials and disposable records. A test that calls an internal helper cannot establish that an HTTP route, MCP adapter or background job always uses that helper.

Verify the saved state separately from execution

Changing an interval and starting an automation are separate operations in this example. The authorized write test checks that running remains false and that the other workspace's record stays unchanged.

Use similarly specific postconditions in an operational integration. After a configuration edit, retrieve the affected record through an authorized read path and compare the fields that should have changed. Check fields that must stay untouched as well. If execution is a separate step, verify its status and result separately.

Read access also deserves a deliberate boundary: a tool can return private configuration without changing it. Decide which records and fields a read operation may reveal.

Applying the distinction in a product workspace

From the SMTasker team: our guide to allowing changes through MCP covers the operator side of this distinction. Read data and Make changes are separate permissions. A request to ask first does not replace the connection's permissions, and write-enabled access can permit supported edits. The guide also explains checking saved settings before running an automation.

The fictional tools and scope names above are not a description of that product's internal implementation. They give developers a small model for testing the access boundary behind their own tool handlers.

Keep the human review clear, enforce the caller's authority in the service, and inspect the resulting state. Each step answers a different question about the same operation.

Which permission check has been hardest to enforce in your MCP integration? Leave your questions or a small example in the comments.