feat: add landlock based access restriction functionality

Landlock is a kernel API for unprivileged access control. We take
advantage of it to limit where unblob can write to and read from on the
filesystem. This is a Linux-only feature that won't be enabled on OSX.

For more information, see https://docs.kernel.org/userspace-api/landlock.html

We use Landlock ABI version 2 since it introduced the
LANDLOCK_ACCESS_FS_REFER permission that's required to create hardlinks.

Co-authored-by: Quentin Kaiser <quentin.kaiser@onekey.com>

Author: vlaci

Cargo.lock

@@ -12,7 +12,12 @@ crate-type = [
 ]
 
 [dependencies]
+log = "0.4.18"
 pyo3 = "0.18.3"
+pyo3-log = "0.8.1"
+
+[target.'cfg(target_os = "linux")'.dependencies]
+landlock = "0.2.0"
 
 [dev-dependencies]
 approx = "0.5.0"

Cargo.toml

@@ -93,6 +93,7 @@ ignore = [
   "D203",    # one-blank-line-before-class:    D211 (no-blank-line-before-class) is used instead
   "D213",    # multi-line-summary-second-line: D212 (multi-line-summary-first-line) is used instead
   "E501",    # line-too-long:                  Let black handle line length violations
+  "UP007",   # non-pep604-annotation:          Python 3.8 support needs legacy annotations
 ]
 
 [tool.ruff.per-file-ignores]

pyproject.toml

@@ -0,0 +1,18 @@
+import os
+import typing
+
+import typing_extensions
+
+_Path: typing_extensions.TypeAlias = typing.Union[os.PathLike, str]
+
+class AccessFS:
+    @staticmethod
+    def read(access_dir: _Path) -> AccessFS: ...
+    @staticmethod
+    def read_write(access_dir: _Path) -> AccessFS: ...
+    @staticmethod
+    def make_reg(access_dir: _Path) -> AccessFS: ...
+    @staticmethod
+    def make_dir(access_dir: _Path) -> AccessFS: ...
+
+def restrict_access(*args: AccessFS) -> None: ...

python/unblob_native/_native/__init__.pyi

@@ -1,11 +1,15 @@
 pub mod math_tools;
+pub mod sandbox;
 
 use pyo3::prelude::*;
 
 /// Performance-critical functionality
 #[pymodule]
 fn _native(py: Python, m: &PyModule) -> PyResult<()> {
     math_tools::init_module(py, m)?;
+    sandbox::init_module(py, m)?;
+
+    pyo3_log::init();
 
     Ok(())
 }

python/unblob_native/_native/math_tools.pyi renamed to python/unblob_native/math_tools.pyi

@@ -0,0 +1,74 @@
+use landlock::{
+    path_beneath_rules, Access, AccessFs, Ruleset, RulesetAttr, RulesetCreatedAttr, ABI,
+};
+use log;
+
+use std::path::Path;
+
+use crate::sandbox::AccessFS;
+
+impl AccessFS {
+    fn read(&self) -> Option<&Path> {
+        if let Self::Read(path) = self {
+            Some(path)
+        } else {
+            None
+        }
+    }
+
+    fn read_write(&self) -> Option<&Path> {
+        if let Self::ReadWrite(path) = self {
+            Some(path)
+        } else {
+            None
+        }
+    }
+
+    fn make_reg(&self) -> Option<&Path> {
+        if let Self::MakeReg(path) = self {
+            Some(path)
+        } else {
+            None
+        }
+    }
+
+    fn make_dir(&self) -> Option<&Path> {
+        if let Self::MakeDir(path) = self {
+            Some(path)
+        } else {
+            None
+        }
+    }
+}
+
+pub fn restrict_access(access_rules: &[AccessFS]) -> Result<(), Box<dyn std::error::Error>> {
+    let abi = ABI::V2;
+
+    let read_only: Vec<&Path> = access_rules.iter().filter_map(AccessFS::read).collect();
+
+    let read_write: Vec<&Path> = access_rules
+        .iter()
+        .filter_map(AccessFS::read_write)
+        .collect();
+
+    let create_file: Vec<&Path> = access_rules.iter().filter_map(AccessFS::make_reg).collect();
+
+    let create_directory: Vec<&Path> = access_rules.iter().filter_map(AccessFS::make_dir).collect();
+
+    let status = Ruleset::new()
+        .handle_access(AccessFs::from_all(abi))?
+        .create()?
+        .add_rules(path_beneath_rules(read_write, AccessFs::from_all(abi)))?
+        .add_rules(path_beneath_rules(create_file, AccessFs::MakeReg))?
+        .add_rules(path_beneath_rules(create_directory, AccessFs::MakeDir))?
+        .add_rules(path_beneath_rules(read_only, AccessFs::from_read(abi)))?
+        .restrict_self()?;
+
+    log::info!(
+        "Activated FS access restrictions; rules={:?}, status={:?}",
+        access_rules,
+        status.ruleset
+    );
+
+    Ok(())
+}

python/unblob_native/sandbox.pyi

@@ -0,0 +1,76 @@
+#[cfg_attr(target_os = "linux", path = "linux.rs")]
+#[cfg_attr(not(target_os = "linux"), path = "unsupported.rs")]
+mod sandbox_impl;
+
+use pyo3::{create_exception, exceptions::PyException, prelude::*, types::PyTuple};
+use std::path::PathBuf;
+
+#[derive(Clone, Debug)]
+pub enum AccessFS {
+    Read(PathBuf),
+    ReadWrite(PathBuf),
+    MakeReg(PathBuf),
+    MakeDir(PathBuf),
+}
+
+/// Enforces access restrictions
+#[pyfunction(name = "restrict_access", signature=(*rules))]
+fn py_restrict_access(rules: &PyTuple) -> PyResult<()> {
+    sandbox_impl::restrict_access(
+        &rules
+            .iter()
+            .map(|r| Ok(r.extract::<PyAccessFS>()?.access))
+            .collect::<PyResult<Vec<_>>>()?,
+    )
+    .map_err(|err| SandboxError::new_err(err.to_string()))
+}
+
+create_exception!(unblob_native.sandbox, SandboxError, PyException);
+
+#[pyclass(name = "AccessFS", module = "unblob_native.sandbox")]
+#[derive(Clone)]
+struct PyAccessFS {
+    access: AccessFS,
+}
+
+impl PyAccessFS {
+    fn new(access: AccessFS) -> Self {
+        Self { access }
+    }
+}
+
+#[pymethods]
+impl PyAccessFS {
+    #[staticmethod]
+    fn read(dir: PathBuf) -> Self {
+        Self::new(AccessFS::Read(dir))
+    }
+
+    #[staticmethod]
+    fn read_write(dir: PathBuf) -> Self {
+        Self::new(AccessFS::ReadWrite(dir))
+    }
+
+    #[staticmethod]
+    fn make_reg(dir: PathBuf) -> Self {
+        Self::new(AccessFS::MakeReg(dir))
+    }
+
+    #[staticmethod]
+    fn make_dir(dir: PathBuf) -> Self {
+        Self::new(AccessFS::MakeDir(dir))
+    }
+}
+
+pub fn init_module(py: Python, root_module: &PyModule) -> PyResult<()> {
+    let module = PyModule::new(py, "sandbox")?;
+    module.add_function(wrap_pyfunction!(py_restrict_access, module)?)?;
+    module.add_class::<PyAccessFS>()?;
+
+    root_module.add_submodule(module)?;
+    py.import("sys")?
+        .getattr("modules")?
+        .set_item("unblob_native.sandbox", module)?;
+
+    Ok(())
+}

src/lib.rs

@@ -0,0 +1,9 @@
+use log;
+
+use crate::sandbox::AccessFS;
+
+pub fn restrict_access(_access_rules: &[AccessFS]) -> Result<(), Box<dyn std::error::Error>> {
+    log::warn!("Sandboxing FS access is unavailable on this system");
+
+    Ok(())
+}