Coverage for wasmtime/_wasi.py: 95%

1import ctypes 

2from ctypes import POINTER, c_char, c_char_p, cast 

3from enum import Enum 

4from os import PathLike 

5from typing import Iterable, List, Union 

6 

7from wasmtime import Managed, WasmtimeError 

8 

9from . import _ffi as ffi 

10from ._config import setter_property 

11 

12 

13def _encode_path(path: Union[str, bytes, PathLike]) -> bytes: 

14 if isinstance(path, (bytes, str)): 

15 path2 = path 

16 else: 

17 path2 = path.__fspath__() 

18 if isinstance(path2, bytes): 

19 return path2 

20 return path2.encode('utf8') 

21 

22class DirPerms(Enum): 

23 READ_ONLY = ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_READ.value 

24 WRITE_ONLY = ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_WRITE.value 

25 READ_WRITE = ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_READ.value | ffi.wasi_dir_perms_flags.WASMTIME_WASI_DIR_PERMS_WRITE.value 

26 

27class FilePerms(Enum): 

28 READ_ONLY = ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_READ.value 

29 WRITE_ONLY = ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_WRITE.value 

30 READ_WRITE = ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_READ.value | ffi.wasi_file_perms_flags.WASMTIME_WASI_FILE_PERMS_WRITE.value 

31 

32class WasiConfig(Managed["ctypes._Pointer[ffi.wasi_config_t]"]): 

33 

34 def __init__(self) -> None: 

35 self._set_ptr(ffi.wasi_config_new()) 

36 

37 def _delete(self, ptr: "ctypes._Pointer[ffi.wasi_config_t]") -> None: 

38 ffi.wasi_config_delete(ptr) 

39 

40 @setter_property 

41 def argv(self, argv: List[str]) -> None: 

42 """ 

43 Explicitly configure the `argv` for this WASI configuration 

44 """ 

45 ptrs = to_char_array(argv) 

46 if not ffi.wasi_config_set_argv(self.ptr(), len(argv), ptrs): 

47 raise WasmtimeError("failed to configure argv") 

48 

49 def inherit_argv(self) -> None: 

50 ffi.wasi_config_inherit_argv(self.ptr()) 

51 

52 @setter_property 

53 def env(self, pairs: Iterable[Iterable]) -> None: 

54 """ 

55 Configure environment variables to be returned for this WASI 

56 configuration. 

57 

58 The `pairs` provided must be an iterable list of key/value pairs of 

59 environment variables. 

60 """ 

61 names = [] 

62 values = [] 

63 for name, value in pairs: 

64 names.append(name) 

65 values.append(value) 

66 name_ptrs = to_char_array(names) 

67 value_ptrs = to_char_array(values) 

68 if not ffi.wasi_config_set_env(self.ptr(), len(names), name_ptrs, value_ptrs): 

69 raise WasmtimeError("failed to configure environment") 

70 

71 def inherit_env(self) -> None: 

72 """ 

73 Configures the environment variables available within WASI to be those 

74 in this own process's environment. All environment variables are 

75 inherited. 

76 """ 

77 ffi.wasi_config_inherit_env(self.ptr()) 

78 

79 @setter_property 

80 def stdin_file(self, path: Union[str, bytes, PathLike]) -> None: 

81 """ 

82 Configures a file to be used as the stdin stream of this WASI 

83 configuration. 

84 

85 Reads of the stdin stream will read the path specified. 

86 

87 The file must already exist on the filesystem. If it cannot be 

88 opened then `WasmtimeError` is raised. 

89 """ 

90 

91 res = ffi.wasi_config_set_stdin_file( 

92 self.ptr(), c_char_p(_encode_path(path))) 

93 if not res: 

94 raise WasmtimeError("failed to set stdin file") 

95 

96 def inherit_stdin(self) -> None: 

97 """ 

98 Configures this own process's stdin to be used as the WASI program's 

99 stdin. 

100 

101 Reads of the stdin stream will read this process's stdin. 

102 """ 

103 ffi.wasi_config_inherit_stdin(self.ptr()) 

104 

105 @setter_property 

106 def stdout_file(self, path: str) -> None: 

107 """ 

108 Configures a file to be used as the stdout stream of this WASI 

109 configuration. 

110 

111 Writes to stdout will be written to the file specified. 

112 

113 The file specified will be created if it doesn't exist, or truncated if 

114 it already exists. It must be available to open for writing. If it 

115 cannot be opened for writing then `WasmtimeError` is raised. 

116 """ 

117 res = ffi.wasi_config_set_stdout_file( 

118 self.ptr(), c_char_p(_encode_path(path))) 

119 if not res: 

120 raise WasmtimeError("failed to set stdout file") 

121 

122 def inherit_stdout(self) -> None: 

123 """ 

124 Configures this own process's stdout to be used as the WASI program's 

125 stdout. 

126 

127 Writes to stdout stream will write to this process's stdout. 

128 """ 

129 ffi.wasi_config_inherit_stdout(self.ptr()) 

130 

131 @setter_property 

132 def stderr_file(self, path: str) -> None: 

133 """ 

134 Configures a file to be used as the stderr stream of this WASI 

135 configuration. 

136 

137 Writes to stderr will be written to the file specified. 

138 

139 The file specified will be created if it doesn't exist, or truncated if 

140 it already exists. It must be available to open for writing. If it 

141 cannot be opened for writing then `WasmtimeError` is raised. 

142 """ 

143 res = ffi.wasi_config_set_stderr_file( 

144 self.ptr(), c_char_p(_encode_path(path))) 

145 if not res: 

146 raise WasmtimeError("failed to set stderr file") 

147 

148 def inherit_stderr(self) -> None: 

149 """ 

150 Configures this own process's stderr to be used as the WASI program's 

151 stderr. 

152 

153 Writes to stderr stream will write to this process's stderr. 

154 """ 

155 ffi.wasi_config_inherit_stderr(self.ptr()) 

156 

157 def preopen_dir(self, path: str, guest_path: str, dir_perms: DirPerms = DirPerms.READ_WRITE, file_perms: FilePerms = FilePerms.READ_WRITE) -> None: 

158 """ 

159 Allows the WASI program to access the directory at `path` using the 

160 path `guest_path` within the WASI program. 

161 

162 `dir_perms` specifies the permissions that wasm will have to operate on 

163 `guest_path`. This can be used, for example, to provide readonly access to a 

164 directory. 

165 

166 `file_perms` specifies the maximum set of permissions that can be used for 

167 any file in this directory. 

168 """ 

169 path_ptr = c_char_p(path.encode('utf-8')) 

170 guest_path_ptr = c_char_p(guest_path.encode('utf-8')) 

171 if not ffi.wasi_config_preopen_dir(self.ptr(), path_ptr, guest_path_ptr, dir_perms.value, file_perms.value): 

172 raise WasmtimeError('failed to add preopen dir') 

173 

174 

175def to_char_array(strings: List[str]) -> "ctypes._Pointer[ctypes._Pointer[c_char]]": 

176 ptrs = (c_char_p * len(strings))() 

177 for i, s in enumerate(strings): 

178 ptrs[i] = c_char_p(s.encode('utf-8')) 

179 return cast(ptrs, POINTER(POINTER(c_char)))