Robotics and ROS 2 authorization

import { createTheAuth } from '@glinr/theauth';
import { ros2Action, ros2Service, ros2Topic } from '@glinr/theauth/robotics';

const theauth = await createTheAuth({
  database: { provider: 'sqlite-native', url: './theauth.db' },
  agents: { enabled: true },
});

const robot = await theauth.agent.create({
  ownerId: 'user-1',
  name: 'robot1-nav',
  type: 'autonomous',
  permissions: [
    { resource: ros2Topic('/scan'), actions: ['subscribe'] },
    { resource: ros2Topic('/odom'), actions: ['publish'] },
    { resource: ros2Service('/reset_costmap'), actions: ['call'] },
    { resource: ros2Action('/navigate_to_pose'), actions: ['execute'] },
    // Hazardous: a person approves each use.
    {
      resource: ros2Topic('/cmd_vel'),
      actions: ['publish'],
      constraints: { requireApproval: true },
    },
  ],
});

theauth ros2 export \
  --robot <agent-id> --server https://auth.example.com/api/auth \
  --enclave /fleet/robot1/nav --out ./robot1-nav

# Planner and drivers: gated permissions left out (default).
theauth ros2 export --robot <planner-agent> --enclave /fleet/robot1/planner --out ./planner

# The gate: gated permissions granted, because the gate is where they are enforced.
theauth ros2 export --robot <gate-agent> --enclave /fleet/robot1/gate --gated include --out ./gate

export ROS_SECURITY_KEYSTORE=~/keystore
ros2 security create_keystore "$ROS_SECURITY_KEYSTORE"          # once
ros2 security create_enclave "$ROS_SECURITY_KEYSTORE" /fleet/robot1/nav
ros2 security create_permission "$ROS_SECURITY_KEYSTORE" /fleet/robot1/nav ./robot1-nav/policy.xml

openssl smime -sign -text -in permissions.xml -out permissions.p7s \
  -signer permissions_ca.cert.pem -inkey permissions_ca.key.pem -md sha256
openssl smime -sign -text -in governance.xml -out governance.p7s \
  -signer permissions_ca.cert.pem -inkey permissions_ca.key.pem -md sha256

# Check a signature:
openssl smime -verify -in permissions.p7s -CAfile permissions_ca.cert.pem -out /dev/null

export ROS_SECURITY_ENABLE=true
export ROS_SECURITY_STRATEGY=Enforce
export ROS_SECURITY_KEYSTORE=~/keystore
ros2 run my_pkg nav_node --ros-args --enclave /fleet/robot1/nav

import { guardRobotCommand, ros2Topic } from '@glinr/theauth/robotics';

const options = {
  baseUrl: 'https://auth.example.com/api/auth',
  token: () => process.env.THEAUTH_ROBOT_TOKEN ?? '',
  timeoutMs: 1000,
  // Exact resources that run without asking. Software stop commands only.
  safetyStops: [ros2Topic('/e_stop')],
};

const result = await guardRobotCommand(
  options,
  { action: 'publish', resource: ros2Topic('/cmd_vel') },
  () => publishVelocity(0.4),
);

if (!result.ok) {
  // result.code: no_permission, approval_required, step_up_required, guard_error, ...
  // result.approvalRequired is true when a person (or a step up) would let it through.
  logRefusal(result.code, result.reason);
}

const req = await theauth.approval.request({
  agentId: robot.id,
  userId: fleetOwnerId,
  action: "publish",
  resource: "ros2:/arm/cmd_vel",
  arguments: { linear: 0.4 },
  requiredApprovals: 2,
});

await theauth.approval.approve(req.id, "operator-a"); // still pending
await theauth.approval.approve(req.id, "operator-b"); // approved

// The gate passes the id on the retry. It works once, for these arguments only.
const result = await theauth.authorize(robot.id, {
  action: "publish",
  resource: "ros2:/arm/cmd_vel",
  arguments: { linear: 0.4 },
  approvalId: req.id,
});

import json
import os
import time
import urllib.error
import urllib.request

import rclpy
from geometry_msgs.msg import Twist
from rclpy.callback_groups import ReentrantCallbackGroup
from rclpy.executors import MultiThreadedExecutor
from rclpy.node import Node

AUTHORIZE_URL = os.environ["THEAUTH_URL"].rstrip("/") + "/authorize/token"
TOKEN = os.environ["THEAUTH_ROBOT_TOKEN"]
TIMEOUT_S = float(os.environ.get("THEAUTH_TIMEOUT_S", "1.0"))
# How long an "allowed" answer is reused. 0 asks every time. A revoked robot keeps moving
# for up to this long, so keep it short.
ALLOW_TTL_S = float(os.environ.get("THEAUTH_ALLOW_TTL_S", "0.5"))


def authorize(action, resource):
    """Return (allowed, code). Any failure is a refusal."""
    body = json.dumps({"action": action, "resource": resource}).encode()
    request = urllib.request.Request(
        AUTHORIZE_URL,
        data=body,
        method="POST",
        headers={"content-type": "application/json", "authorization": "Bearer " + TOKEN},
    )
    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT_S) as response:
            data = json.load(response).get("data", {})
    except urllib.error.HTTPError as err:
        try:
            data = json.load(err).get("data", {})
        except ValueError:
            data = {}
    except Exception:
        return False, "guard_error"
    return data.get("allowed") is True, data.get("code", "guard_error")


def is_stop(msg):
    """A command that only asks for zero motion needs no permission."""
    return all(v == 0.0 for v in (
        msg.linear.x, msg.linear.y, msg.linear.z,
        msg.angular.x, msg.angular.y, msg.angular.z,
    ))


class CommandGate(Node):
    def __init__(self):
        super().__init__("command_gate")
        group = ReentrantCallbackGroup()
        self.pub = self.create_publisher(Twist, "/cmd_vel", 10)
        self.sub = self.create_subscription(
            Twist, "/cmd_vel_request", self.on_request, 10, callback_group=group
        )
        self.allowed_until = 0.0

    def on_request(self, msg):
        if is_stop(msg):
            self.pub.publish(msg)
            return
        now = time.monotonic()
        if now >= self.allowed_until:
            allowed, code = authorize("publish", "ros2:/cmd_vel")
            if not allowed:
                self.get_logger().warn("cmd_vel refused: " + code)
                return
            self.allowed_until = now + ALLOW_TTL_S
        self.pub.publish(msg)


def main():
    rclpy.init()
    node = CommandGate()
    executor = MultiThreadedExecutor()
    executor.add_node(node)
    executor.spin()


if __name__ == "__main__":
    main()

#include <chrono>
#include <geometry_msgs/msg/twist.hpp>
#include <rclcpp/rclcpp.hpp>

// Returns true only for HTTP 200 with {"data":{"allowed":true}}.
// Returns false on timeout, transport error, bad JSON, or any other status.
bool theauth_allows(const std::string &action, const std::string &resource);

class CommandGate : public rclcpp::Node {
 public:
  CommandGate() : Node("command_gate") {
    pub_ = create_publisher<geometry_msgs::msg::Twist>("/cmd_vel", 10);
    sub_ = create_subscription<geometry_msgs::msg::Twist>(
        "/cmd_vel_request", 10,
        [this](geometry_msgs::msg::Twist::ConstSharedPtr msg) { on_request(*msg); });
  }

 private:
  static bool is_stop(const geometry_msgs::msg::Twist &m) {
    return m.linear.x == 0 && m.linear.y == 0 && m.linear.z == 0 &&
           m.angular.x == 0 && m.angular.y == 0 && m.angular.z == 0;
  }

  void on_request(const geometry_msgs::msg::Twist &msg) {
    if (!is_stop(msg) && !theauth_allows("publish", "ros2:/cmd_vel")) {
      RCLCPP_WARN(get_logger(), "cmd_vel refused");
      return;
    }
    pub_->publish(msg);
  }

  rclcpp::Publisher<geometry_msgs::msg::Twist>::SharedPtr pub_;
  rclcpp::Subscription<geometry_msgs::msg::Twist>::SharedPtr sub_;
};

theauth ros2 diff before.json after.json

# Host. perms: /chatter publish plus a gated /cmd_vel (requireApproval) for the talker.
theauth ros2 export --permissions talker.json --enclave /w11/talker --out out-talker
#   9 rules exported, 1 gated, 0 skipped.
#     gated  ros2:/cmd_vel [publish] left out: requireApproval
theauth ros2 export --permissions listener.json --enclave /w11/listener --out out-listener
theauth ros2 export --permissions driver.json   --enclave /w11/driver   --out out-driver   # subscribes /cmd_vel
theauth ros2 export --permissions gate.json     --enclave /w11/gate     --out out-gate --gated include

# Container.
ros2 security create_keystore "$ROS_SECURITY_KEYSTORE"
for e in talker listener gate driver; do
  ros2 security create_enclave   "$ROS_SECURITY_KEYSTORE" /w11/$e
  ros2 security create_permission "$ROS_SECURITY_KEYSTORE" /w11/$e /work/out-$e/policy.xml
done
export ROS_SECURITY_ENABLE=true ROS_SECURITY_STRATEGY=Enforce

[SECURITY Error] rt/secret topic not found in allow rule. (./src/cpp/security/accesscontrol/Permissions.cpp:1245) -> Function check_create_datareader

rt/rosout topic not found in allow rule ... check_create_datawriter
rq/talker/describe_parametersRequest topic not found in allow rule ... check_create_datareader
rt/parameter_events topic not found in allow rule ... check_create_datawriter

/fleet        refused
/fleet/a      allowed
/fleet/a/b    allowed
/fleet/a/b/c  allowed
/other        refused