use {
crate::NewInterpreterError,
oxidized_importer::{PackedResourcesSource, PythonResourcesState},
pyo3::ffi as pyffi,
python_packaging::interpreter::{
MemoryAllocatorBackend, MultiprocessingStartMethod, PythonInterpreterConfig,
PythonInterpreterProfile, TerminfoResolution,
},
std::{
ffi::{CString, OsString},
ops::Deref,
path::PathBuf,
},
};
#[cfg(feature = "serialization")]
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug)]
pub struct ExtensionModule {
pub name: CString,
pub init_func: unsafe extern "C" fn() -> *mut pyffi::PyObject,
}
#[derive(Clone, Debug)]
#[cfg_attr(feature = "serialization", derive(Deserialize, Serialize))]
#[cfg_attr(feature = "serialization", serde(default))]
pub struct OxidizedPythonInterpreterConfig<'a> {
pub exe: Option<PathBuf>,
pub origin: Option<PathBuf>,
pub interpreter_config: PythonInterpreterConfig,
pub allocator_backend: MemoryAllocatorBackend,
pub allocator_raw: bool,
pub allocator_mem: bool,
pub allocator_obj: bool,
pub allocator_pymalloc_arena: bool,
pub allocator_debug: bool,
pub set_missing_path_configuration: bool,
pub oxidized_importer: bool,
pub filesystem_importer: bool,
#[cfg_attr(feature = "serialization", serde(skip))]
pub packed_resources: Vec<PackedResourcesSource<'a>>,
#[cfg_attr(feature = "serialization", serde(skip))]
pub extra_extension_modules: Option<Vec<ExtensionModule>>,
pub argv: Option<Vec<OsString>>,
pub argvb: bool,
pub multiprocessing_auto_dispatch: bool,
pub multiprocessing_start_method: MultiprocessingStartMethod,
pub sys_frozen: bool,
pub sys_meipass: bool,
pub terminfo_resolution: TerminfoResolution,
pub tcl_library: Option<PathBuf>,
pub write_modules_directory_env: Option<String>,
}
impl<'a> Default for OxidizedPythonInterpreterConfig<'a> {
fn default() -> Self {
Self {
exe: None,
origin: None,
interpreter_config: PythonInterpreterConfig {
profile: PythonInterpreterProfile::Python,
..PythonInterpreterConfig::default()
},
allocator_backend: MemoryAllocatorBackend::Default,
allocator_raw: true,
allocator_mem: false,
allocator_obj: false,
allocator_pymalloc_arena: false,
allocator_debug: false,
set_missing_path_configuration: true,
oxidized_importer: false,
filesystem_importer: true,
packed_resources: vec![],
extra_extension_modules: None,
argv: None,
argvb: false,
multiprocessing_auto_dispatch: true,
multiprocessing_start_method: MultiprocessingStartMethod::Auto,
sys_frozen: false,
sys_meipass: false,
terminfo_resolution: TerminfoResolution::Dynamic,
tcl_library: None,
write_modules_directory_env: None,
}
}
}
impl<'a> OxidizedPythonInterpreterConfig<'a> {
pub fn resolve(
self,
) -> Result<ResolvedOxidizedPythonInterpreterConfig<'a>, NewInterpreterError> {
let argv = if let Some(args) = self.argv {
Some(args)
} else if self.interpreter_config.argv.is_some() {
None
} else {
Some(std::env::args_os().collect::<Vec<_>>())
};
let exe = if let Some(exe) = self.exe {
exe
} else {
std::env::current_exe()
.map_err(|_| NewInterpreterError::Simple("could not obtain current executable"))?
};
let exe = dunce::canonicalize(exe)
.map_err(|_| NewInterpreterError::Simple("could not obtain current executable path"))?;
let origin = if let Some(origin) = self.origin {
origin
} else {
exe.parent()
.ok_or(NewInterpreterError::Simple(
"unable to obtain current executable parent directory",
))?
.to_path_buf()
};
let origin_string = origin.display().to_string();
let packed_resources = self
.packed_resources
.into_iter()
.map(|entry| match entry {
PackedResourcesSource::Memory(_) => entry,
PackedResourcesSource::MemoryMappedPath(p) => {
PackedResourcesSource::MemoryMappedPath(PathBuf::from(
p.display().to_string().replace("$ORIGIN", &origin_string),
))
}
})
.collect::<Vec<_>>();
let module_search_paths = self
.interpreter_config
.module_search_paths
.as_ref()
.map(|x| {
x.iter()
.map(|p| {
PathBuf::from(p.display().to_string().replace("$ORIGIN", &origin_string))
})
.collect::<Vec<_>>()
});
let tcl_library = self
.tcl_library
.as_ref()
.map(|x| PathBuf::from(x.display().to_string().replace("$ORIGIN", &origin_string)));
Ok(ResolvedOxidizedPythonInterpreterConfig {
inner: Self {
exe: Some(exe),
origin: Some(origin),
interpreter_config: PythonInterpreterConfig {
module_search_paths,
..self.interpreter_config
},
argv,
packed_resources,
tcl_library,
..self
},
})
}
}
pub struct ResolvedOxidizedPythonInterpreterConfig<'a> {
inner: OxidizedPythonInterpreterConfig<'a>,
}
impl<'a> Deref for ResolvedOxidizedPythonInterpreterConfig<'a> {
type Target = OxidizedPythonInterpreterConfig<'a>;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl<'a> TryFrom<OxidizedPythonInterpreterConfig<'a>>
for ResolvedOxidizedPythonInterpreterConfig<'a>
{
type Error = NewInterpreterError;
fn try_from(value: OxidizedPythonInterpreterConfig<'a>) -> Result<Self, Self::Error> {
value.resolve()
}
}
impl<'a> ResolvedOxidizedPythonInterpreterConfig<'a> {
pub fn exe(&self) -> &PathBuf {
self.inner.exe.as_ref().expect("exe should have a value")
}
pub fn origin(&self) -> &PathBuf {
self.inner
.origin
.as_ref()
.expect("origin should have a value")
}
pub fn resolve_sys_argv(&self) -> &[OsString] {
if let Some(args) = &self.inner.argv {
args
} else if let Some(args) = &self.inner.interpreter_config.argv {
args
} else {
panic!("1 of .argv or .interpreter_config.argv should be set")
}
}
pub fn resolve_sys_argvb(&self) -> Vec<OsString> {
if let Some(args) = &self.inner.interpreter_config.argv {
args.clone()
} else if let Some(args) = &self.inner.argv {
args.clone()
} else {
std::env::args_os().collect::<Vec<_>>()
}
}
}
impl<'a, 'config: 'a> TryFrom<&ResolvedOxidizedPythonInterpreterConfig<'config>>
for PythonResourcesState<'a, u8>
{
type Error = NewInterpreterError;
fn try_from(
config: &ResolvedOxidizedPythonInterpreterConfig<'config>,
) -> Result<Self, Self::Error> {
let mut state = Self::default();
state.set_current_exe(config.exe().to_path_buf());
state.set_origin(config.origin().to_path_buf());
for source in &config.packed_resources {
match source {
PackedResourcesSource::Memory(data) => {
state
.index_data(data)
.map_err(NewInterpreterError::Simple)?;
}
PackedResourcesSource::MemoryMappedPath(path) => {
state
.index_path_memory_mapped(path)
.map_err(NewInterpreterError::Dynamic)?;
}
}
}
state
.index_interpreter_builtins()
.map_err(NewInterpreterError::Simple)?;
Ok(state)
}
}
#[cfg(test)]
mod tests {
use {super::*, anyhow::Result};
#[test]
fn test_packed_resources_implicit_origin() -> Result<()> {
let mut config = OxidizedPythonInterpreterConfig::default();
config
.packed_resources
.push(PackedResourcesSource::MemoryMappedPath(PathBuf::from(
"$ORIGIN/lib/packed-resources",
)));
let resolved = config.resolve()?;
assert_eq!(
resolved.packed_resources,
vec![PackedResourcesSource::MemoryMappedPath(
resolved.origin().join("lib/packed-resources")
)]
);
Ok(())
}
#[test]
fn test_packed_resources_explicit_origin() -> Result<()> {
let mut config = OxidizedPythonInterpreterConfig {
origin: Some(PathBuf::from("/other/origin")),
..Default::default()
};
config
.packed_resources
.push(PackedResourcesSource::MemoryMappedPath(PathBuf::from(
"$ORIGIN/lib/packed-resources",
)));
let resolved = config.resolve()?;
assert_eq!(
resolved.packed_resources,
vec![PackedResourcesSource::MemoryMappedPath(PathBuf::from(
"/other/origin/lib/packed-resources"
))]
);
Ok(())
}
}