-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
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 <[email protected]>
- Loading branch information
Showing
10 changed files
with
319 additions
and
3 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,16 @@ | ||
import os | ||
import typing | ||
|
||
_Path = 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: ... |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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(()) | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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(()) | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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(()) | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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(()) | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,64 @@ | ||
import platform | ||
from pathlib import Path | ||
|
||
import pytest | ||
|
||
from unblob_native.sandbox import AccessFS, restrict_access | ||
|
||
FILE_CONTENT = b"HELLO" | ||
|
||
|
||
@pytest.mark.skipif(platform.system() == "Linux", reason="Linux is supported.") | ||
def test_unsupported_platform(): | ||
restrict_access(AccessFS.read("/")) | ||
|
||
|
||
@pytest.fixture(scope="session") | ||
def sandbox_path(tmp_path_factory: pytest.TempPathFactory) -> Path: | ||
sandbox_path = tmp_path_factory.mktemp("sandbox") | ||
|
||
file_path = sandbox_path / "file.txt" | ||
dir_path = sandbox_path / "dir" | ||
link_path = sandbox_path / "link" | ||
|
||
with file_path.open("wb") as f: | ||
assert f.write(FILE_CONTENT) == len(FILE_CONTENT) | ||
|
||
dir_path.mkdir() | ||
link_path.symlink_to(file_path) | ||
|
||
return sandbox_path | ||
|
||
|
||
@pytest.mark.skipif( | ||
platform.system() != "Linux" or platform.machine() != "x86_64", | ||
reason="Only supported on Linux x86-64.", | ||
) | ||
def test_read_sandboxing(request: pytest.FixtureRequest, sandbox_path: Path): | ||
restrict_access( | ||
AccessFS.read("/"), | ||
AccessFS.read(sandbox_path), | ||
# allow pytest caching, coverage, etc... | ||
AccessFS.read_write(request.config.rootpath), | ||
) | ||
|
||
with pytest.raises(PermissionError): | ||
(sandbox_path / "some-dir").mkdir() | ||
|
||
with pytest.raises(PermissionError): | ||
(sandbox_path / "some-file").touch() | ||
|
||
with pytest.raises(PermissionError): | ||
(sandbox_path / "some-link").symlink_to("file.txt") | ||
|
||
for path in sandbox_path.rglob("**/*"): | ||
if path.is_file() or path.is_symlink(): | ||
with path.open("rb") as f: | ||
assert f.read() == FILE_CONTENT | ||
with pytest.raises(PermissionError): | ||
assert path.open("r+") | ||
with pytest.raises(PermissionError): | ||
assert path.unlink() | ||
elif path.is_dir(): | ||
with pytest.raises(PermissionError): | ||
path.rmdir() |