maelys-pty

Level 1 — Line filter with metachar scan

Inspect command lines, project metacharacters into facts, and query an explicit allow.

Choose C or Python in the synchronized tabs. Both versions use the current opaque consumer API: C includes <maelys/datalog.h> and <maelys/datalog_builders.h>; Python imports maelys_datalog. Use the quickstart setup for Maelys Datalog 0.11.1. Scope: this is an integration walkthrough, not a production sandbox. The existing C terminal and syscall snippets are sketches requiring a complete native monitor; Python supplies policy evaluation, not a ptrace or seccomp implementation. Do not treat the line scanner, pathname resolution or these snippets as a security boundary.

Level 1 wraps an interactive shell behind a pseudo-terminal and intercepts every typed line before it reaches the child shell. It scans for shell metacharacters (|, >, ;, `, $VAR) before tokenising, injects them as EDB facts, and evaluates a Datalog policy. If the policy does not derive allow(Cmd), the line is discarded and never forwarded to the shell.

What we are building

  1. User
    types a command
  2. Metachar
    scan before split
  3. maelys-pty
    tokenise + EDB
  4. Policy
    allow / deny
  5. Shell
    bash via slave pty

The problem with naive tokenisation

strtok(line, " \t") breaks on whitespace. It cannot handle:

CODE
git commit -m "feat: add solver"   → strtok splits on space inside quotes
echo "shutdown now"                → tokens[0] = echo, correct — but what
                                     about: shutdown now ?
| rm -rf /                         → tokens[0] = "|", but the pipe is the threat
> /etc/passwd                      → redirect silently ignored by line filter
CMD=rm; $CMD /                     → $CMD never resolved — policy sees "$CMD"

For this demonstration, restrict the accepted grammar to unquoted, whitespace-separated tokens. Quotes, backslashes, ampersands and line breaks deny in both languages. Scan metacharacters before tokenising and inject the remaining operators as EDB facts. The policy then decides whether the presence of a pipe or a variable expansion is itself grounds for denial — independently of what the command token happens to be.

Domain

CODEDeclare the terminal vocabulary
#include <maelys/datalog.h>
#include <maelys/datalog_builders.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>

static const maelys_datalog_predicate_t pty_preds[] = {
  /* EDB: populated per command line */
  { "command",        1, MAELYS_DATALOG_PREDICATE_EDB },
  { "arg",            1, MAELYS_DATALOG_PREDICATE_EDB },
  { "user",           1, MAELYS_DATALOG_PREDICATE_EDB },
  { "cwd",            1, MAELYS_DATALOG_PREDICATE_EDB },
  { "time_of_day",    1, MAELYS_DATALOG_PREDICATE_EDB },
  /* Metacharacters detected in the raw line before tokenising */
  { "shell_metachar", 1, MAELYS_DATALOG_PREDICATE_EDB },

  /* POLICY_FACT */
  { "forbidden_cmd",  1, MAELYS_DATALOG_PREDICATE_POLICY_FACT },
  { "trusted_user",   1, MAELYS_DATALOG_PREDICATE_POLICY_FACT },

  /* IDB + QUERY */
  { "deny_any",  1, MAELYS_DATALOG_PREDICATE_IDB | MAELYS_DATALOG_PREDICATE_QUERY },
  { "allow",     1, MAELYS_DATALOG_PREDICATE_IDB | MAELYS_DATALOG_PREDICATE_QUERY },
  { "deny",      2, MAELYS_DATALOG_PREDICATE_IDB | MAELYS_DATALOG_PREDICATE_QUERY },
};

static const char *const POLICY_ATOMS[] = {
    "shutdown",
    "reboot",
    "poweroff",
    "alice",
    "root",
    "__line__",
    "shell metacharacter in line",
    "command is forbidden by policy",
    "curl",
    "curl is restricted to trusted users"
};
static const maelys_datalog_domain_t pty_domain = {
    "pty_policy", pty_preds, sizeof(pty_preds) / sizeof(pty_preds[0]),
    POLICY_ATOMS, sizeof(POLICY_ATOMS) / sizeof(POLICY_ATOMS[0]),
};

Scanning metacharacters before tokenising

CODEInspect the supported line syntax
enum {
    META_PIPE = 1u, META_REDIRECT = 2u, META_SEMICOLON = 4u,
    META_BACKTICK = 8u, META_VAR = 16u
};

/* Restricted demonstration grammar: unquoted, whitespace-separated tokens. */
static int scan_metacharacters(const char *line, size_t len, unsigned *flags)
{
    *flags = 0;
    for (size_t i = 0; i < len; ++i) {
        switch (line[i]) {
        case '\0': case '&': case '\\': case '\n': case '\r':
        case '\'': case '"': return 0; /* unsupported syntax: deny */
        case '|': *flags |= META_PIPE; break;
        case '>': case '<': *flags |= META_REDIRECT; break;
        case ';': *flags |= META_SEMICOLON; break;
        case '`': *flags |= META_BACKTICK; break;
        case '$': *flags |= META_VAR; break;
        }
    }
    return 1;
}

static maelys_datalog_status_t inject_metachar_facts(
    maelys_datalog_input_edb_t *edb, unsigned flags,
    maelys_datalog_diagnostic_t *diagnostic)
{
    const unsigned bits[] = {META_PIPE, META_REDIRECT, META_SEMICOLON, META_BACKTICK, META_VAR};
    const char *const names[] = {"pipe", "redirect", "semicolon", "backtick", "var_expand"};
    for (size_t i = 0; i < 5; ++i) {
        if (flags & bits[i]) {
            maelys_datalog_status_t rc = MAELYS_DATALOG_ADD_FACT(
                edb, diagnostic, "shell_metachar", names[i]);
            if (rc != MAELYS_DATALOG_STATUS_OK) return rc;
        }
    }
    return MAELYS_DATALOG_STATUS_OK;
}

The Level 1 policy

DATALOG
/* ── Forbidden commands ─────────────────────────────────────── */
forbidden_cmd("shutdown").
forbidden_cmd("reboot").
forbidden_cmd("poweroff").

/* ── Trusted users ──────────────────────────────────────────── */
trusted_user("alice").
trusted_user("root").

/* ── Shell metacharacter lines are always blocked ───────────── */
/* Any line containing a pipe, redirect, semicolon, backtick,
   or unquoted variable expansion is rejected at the line level.
   This is independent of what the command token happens to be. */
deny("__line__", "shell metacharacter in line") :-
    shell_metachar(_).

/* ── Forbidden binaries ─────────────────────────────────────── */
deny(Cmd, "command is forbidden by policy") :-
    command(Cmd),
    forbidden_cmd(Cmd).

/* ── curl restricted to trusted users ──────────────────────── */
deny("curl", "curl is restricted to trusted users") :-
    command("curl"),
    user(U),
    not(trusted_user(U)).

/* ── deny_any / allow ───────────────────────────────────────── */
deny_any(Op) :- deny(Op, Reason).

allow(Cmd) :-
    command(Cmd),
    not(deny_any("__line__")),
    not(deny_any(Cmd)).

The __line__ sentinel is a stable string value for line-level denials — metacharacter checks apply before any command-level evaluation.

Load the policy once

CODELoad the terminal policy
static maelys_datalog_status_t load_pty_policy(
    const char *source, maelys_datalog_session_t **session,
    maelys_datalog_diagnostic_t *diagnostic)
{
    maelys_datalog_status_t rc = maelys_datalog_domain_register(&pty_domain);
    if (rc != MAELYS_DATALOG_STATUS_OK) return rc;
    maelys_datalog_policy_t *policy = NULL;
    rc = maelys_datalog_policy_load_inline("pty_policy", "pty_policy.l1",
        source, strlen(source), &policy, diagnostic);
    if (rc != MAELYS_DATALOG_STATUS_OK) return rc;
    rc = maelys_datalog_session_create(policy, 0u, session);
    (void)maelys_datalog_policy_free(policy);
    return rc;
}

Building the EDB

CODEEvaluate a line without executing it
/* Application writer: return 0 only after successfully recording the decision. */
typedef int (*decision_writer_t)(const char *command, const char *reason);

static int evaluate_line(maelys_datalog_session_t *session,
    const char *line, size_t len, const char *user, int confirmed,
    decision_writer_t record_decision)
{
    if (!line || !user || !user[0] || len >= 4096u) return 0;
    unsigned flags = 0;
    if (!scan_metacharacters(line, len, &flags)) return 0;
    char copy[4096];
    memcpy(copy, line, len);
    copy[len] = '\0';
    char *tokens[63];
    size_t ntok = 0;
    for (char *p = strtok(copy, " \t"); p; p = strtok(NULL, " \t")) {
        if (ntok == 63u) return 0;
        tokens[ntok++] = p;
    }
    if (ntok == 0u) return 0; /* no command to authorize */

    maelys_datalog_input_edb_t *edb = NULL;
    maelys_datalog_result_t *result = NULL;
    maelys_datalog_diagnostic_t diagnostic = MAELYS_DATALOG_DIAGNOSTIC_INIT;
    int allowed = 0;
    maelys_datalog_status_t rc = maelys_datalog_input_edb_create(&edb);
    if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
    rc = inject_metachar_facts(edb, flags, &diagnostic);
    if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
    rc = MAELYS_DATALOG_ADD_FACTS(edb, &diagnostic,
        MAELYS_DATALOG_FACT("command", tokens[0]),
        MAELYS_DATALOG_FACT("user", user));
    if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
    for (size_t i = 1; i < ntok; ++i) {
        rc = MAELYS_DATALOG_ADD_FACT(edb, &diagnostic, "arg", tokens[i]);
        if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
    }
    if (confirmed) {
        rc = MAELYS_DATALOG_ADD_FACT(edb, &diagnostic, "confirmed", tokens[0]);
        if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
    }
    rc = maelys_datalog_session_solve_edb(session, edb, &result, &diagnostic);
    if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
    rc = MAELYS_DATALOG_QUERY(result, &allowed, "allow", tokens[0]);
    if (rc != MAELYS_DATALOG_STATUS_OK || allowed || !record_decision) goto cleanup;
    /* Level 2 only: failed or oversized enumeration must deny. */
    maelys_datalog_fact_view_t rows[64];
    size_t count = 0;
    rc = maelys_datalog_result_enumerate(result, "allow_with_log", 2u, rows, 64u, &count);
    if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
    for (size_t i = 0; i < count; ++i) {
        const char *command = NULL, *reason = NULL;
        size_t length = 0;
        rc = maelys_datalog_result_symbol_text(result, rows[i].terms[0].as.symbol_id,
                                              &command, &length);
        if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
        rc = maelys_datalog_result_symbol_text(result, rows[i].terms[1].as.symbol_id,
                                              &reason, &length);
        if (rc != MAELYS_DATALOG_STATUS_OK) goto cleanup;
        if (strcmp(command, tokens[0]) == 0) {
            allowed = record_decision(command, reason) == 0;
            break;
        }
    }
cleanup:
    if (rc != MAELYS_DATALOG_STATUS_OK)
        fprintf(stderr, "maelys-pty: %s: %s\n",
                maelys_datalog_status_name(rc), diagnostic.message);
    if (result) (void)maelys_datalog_result_free(result);
    (void)maelys_datalog_input_edb_free(edb);
    return rc == MAELYS_DATALOG_STATUS_OK && allowed;
}

Level 1 in action

CODE
alice@host:~$ shutdown now
maelys-pty: 'shutdown' denied by policy

alice@host:~$ echo hello | grep hello
maelys-pty: '__line__' denied by policy   ← pipe detected before tokenising

alice@host:~$ git commit -m "feat: solver"
maelys-pty: denied before solving — quoted syntax is unsupported in this demo

alice@host:~$ CMD=rm; $CMD /
maelys-pty: '__line__' denied by policy   ← semicolon + $VAR both detected

alice@host:~$ git push origin feature/x
→ allowed: no metacharacters, not forbidden

What you have learned

Keep this model

Inspect command structure before tokenization
  • Scan the original line

    Record shell metacharacters before splitting the input into command tokens. Pipes, redirections and substitutions must not disappear during parsing.

  • Evaluate explicit command and line evidence

    The wrapper builds fresh facts for the line and checks the relevant denials and explicit allow before forwarding it.

  • Know the scanner boundary

    This example accepts a deliberately restricted grammar. It is not a complete shell parser or a production sandbox, and later execution may need native monitoring.

Continue