feat: add landlock based access restriction functionality

Cargo.toml

@@ -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"

src/lib.rs

@@ -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(())
 }

src/sandbox/linux.rs

@@ -0,0 +1,72 @@
+use landlock::{
+ path_beneath_rules, Access, AccessFs, Ruleset, RulesetAttr, RulesetCreatedAttr, ABI,
+};
+use log;
+
+use crate::sandbox::AccessFS;
+
+impl AccessFS {
+ fn read(&self) -> Option<&str> {
+ if let Self::Read(path) = self {
+ Some(path)
+ } else {
+ None
+ }
+ }
+
+ fn read_write(&self) -> Option<&str> {
+ if let Self::ReadWrite(path) = self {
+ Some(path)
+ } else {
+ None
+ }
+ }
+
+ fn make_reg(&self) -> Option<&str> {
+ if let Self::MakeReg(path) = self {
+ Some(path)
+ } else {
+ None
+ }
+ }
+
+ fn make_dir(&self) -> Option<&str> {
+ 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::V1;
+
+ let read_only: Vec<&str> = access_rules.iter().filter_map(AccessFS::read).collect();
+
+ let read_write: Vec<&str> = access_rules
+ .iter()
+ .filter_map(AccessFS::read_write)
+ .collect();
+
+ let create_file: Vec<&str> = access_rules.iter().filter_map(AccessFS::make_reg).collect();
+
+ let create_directory: Vec<&str> = 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_only, AccessFs::from_read(abi)))?
+ .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))?
+ .restrict_self()?;
+
+ log::info!(
+ "Activated FS access restrictions; rules={:?}, status={:?}",
+ access_rules,
+ status.ruleset
+ );
+
+ Ok(())
+}

src/sandbox/mod.rs

@@ -0,0 +1,80 @@
+#[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};
+
+#[derive(Clone, Debug)]
+pub enum AccessFS {
+ Read(String),
+ ReadWrite(String),
+ MakeReg(String),
+ MakeDir(String),
+}
+
+/// 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: String) -> Self {
+ Self::new(AccessFS::Read(dir))
+ }
+
+ #[staticmethod]
+ fn read_write(dir: String) -> Self {
+ Self::new(AccessFS::ReadWrite(dir))
+ }
+
+ #[staticmethod]
+ fn make_reg(dir: String) -> Self {
+ Self::new(AccessFS::MakeReg(dir))
+ }
+
+ #[staticmethod]
+ fn make_dir(dir: String) -> 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)?;
+
+ let sys = PyModule::import(py, "sys")?;
+ let modules = sys.getattr("modules")?;
+ modules.call_method(
+ "__setitem__",
+ ("unblob_native.sandbox".to_string(), module),
+ None,
+ )?;
+
+ Ok(())
+}

src/sandbox/unsupported.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(())
+}