1 | """Manage Network Application files.""" |
||
2 | 1 | import json |
|
3 | 1 | import logging |
|
4 | 1 | import re |
|
5 | 1 | import shutil |
|
6 | 1 | from pathlib import Path |
|
7 | |||
8 | 1 | from kytos.core.napps import NApp |
|
9 | |||
10 | 1 | LOG = logging.getLogger(__name__) |
|
11 | |||
12 | |||
13 | 1 | class NAppsManager: |
|
14 | """Deal with NApps at filesystem level and ask Kytos to (un)load NApps.""" |
||
15 | |||
16 | 1 | def __init__(self, controller=None, base_path=None): |
|
17 | """Need the controller for configuration paths and (un)loading NApps. |
||
18 | |||
19 | Args: |
||
20 | controller (kytos.Controller): Controller to (un)load NApps. |
||
21 | base_path (pathlib.Path): base path for enabled NApps. |
||
22 | This will be supported while kytos-utils still imports |
||
23 | kytos.core directly, and may be removed when it calls Kytos' |
||
24 | Web API. |
||
25 | """ |
||
26 | 1 | self._controller = controller |
|
27 | |||
28 | 1 | if base_path: |
|
29 | self._enabled_path = base_path |
||
30 | else: |
||
31 | 1 | self._config = controller.options |
|
32 | 1 | self._enabled_path = Path(self._config.napps) |
|
33 | |||
34 | 1 | self._installed_path = self._enabled_path / '.installed' |
|
35 | |||
36 | 1 | def install(self, napp_uri, enable=True): |
|
37 | """Install and enable a NApp from its repository. |
||
38 | |||
39 | By default, install procedure will also enable the NApp. If you only |
||
40 | want to install and keep NApp disabled, please use enable=False. |
||
41 | """ |
||
42 | napp = NApp.create_from_uri(napp_uri) |
||
43 | |||
44 | if napp in self.get_all_napps(): |
||
45 | LOG.warning("Unable to install NApp %s. Already installed.", napp) |
||
46 | return False |
||
47 | |||
48 | if not napp.repository: |
||
49 | napp.repository = self._controller.options.napps_repositories[0] |
||
50 | |||
51 | pkg_folder = None |
||
52 | try: |
||
53 | pkg_folder = napp.download() |
||
54 | napp_folder = self._find_napp(napp, pkg_folder) |
||
55 | dst = self._installed_path / napp.username / napp.name |
||
56 | self._create_module(dst.parent) |
||
57 | shutil.move(str(napp_folder), str(dst)) |
||
58 | finally: |
||
59 | if pkg_folder and pkg_folder.exists(): |
||
60 | shutil.rmtree(str(pkg_folder)) |
||
61 | |||
62 | LOG.info("New NApp installed: %s", napp) |
||
63 | |||
64 | napp = NApp.create_from_json(dst/'kytos.json') |
||
65 | for uri in napp.napp_dependencies: |
||
66 | self.install(uri, enable) |
||
67 | |||
68 | if enable: |
||
69 | return self.enable(napp.username, napp.name) |
||
70 | |||
71 | return True |
||
72 | |||
73 | 1 | def uninstall(self, username, napp_name): |
|
74 | """Remove a NApp from filesystem, if existent.""" |
||
75 | napp_id = "{}/{}".format(username, napp_name) |
||
76 | |||
77 | if self.is_enabled(username, napp_name): |
||
78 | LOG.warning("Unable to uninstall NApp %s. NApp currently in use.", |
||
79 | napp_id) |
||
80 | return False |
||
81 | |||
82 | new_manager = NewNAppManager(self._installed_path) |
||
83 | napp = new_manager.napps[napp_id] |
||
84 | deps = napp.napp_dependencies |
||
85 | |||
86 | if deps and napp.meta: |
||
87 | LOG.info('Uninstalling Meta-NApp %s dependencies: %s', napp, deps) |
||
88 | for uri in deps: |
||
89 | username, napp_name = self.get_napp_fullname_from_uri(uri) |
||
90 | self.uninstall(username, napp_name) |
||
91 | |||
92 | if self.is_installed(username, napp_name): |
||
93 | installed = self._installed_path / napp_id |
||
94 | if installed.is_symlink(): |
||
95 | installed.unlink() |
||
96 | else: |
||
97 | shutil.rmtree(str(installed)) |
||
98 | LOG.info("NApp uninstalled: %s", napp_id) |
||
99 | else: |
||
100 | LOG.warning("Unable to uninstall NApp %s. Already uninstalled.", |
||
101 | napp_id) |
||
102 | return True |
||
103 | |||
104 | 1 | def enable(self, username, napp_name): |
|
105 | """Enable a NApp if not already enabled.""" |
||
106 | napp_id = "{}/{}".format(username, napp_name) |
||
107 | |||
108 | enabled = self._enabled_path / napp_id |
||
109 | installed = self._installed_path / napp_id |
||
110 | |||
111 | new_manager = NewNAppManager(self._installed_path) |
||
112 | napp = new_manager.napps[napp_id] |
||
113 | deps = napp.napp_dependencies |
||
114 | |||
115 | if deps and napp.meta: |
||
116 | LOG.info('Enabling Meta-NApp %s dependencies: %s', napp, deps) |
||
117 | for uri in deps: |
||
118 | username, napp_name = self.get_napp_fullname_from_uri(uri) |
||
119 | self.enable(username, napp_name) |
||
120 | |||
121 | if not installed.is_dir(): |
||
122 | LOG.error("Failed to enable NApp %s. NApp not installed.", napp_id) |
||
123 | elif not enabled.exists(): |
||
124 | self._create_module(enabled.parent) |
||
125 | try: |
||
126 | # Create symlink |
||
127 | enabled.symlink_to(installed) |
||
128 | if self._controller is not None: |
||
129 | self._controller.load_napp(username, napp_name) |
||
130 | LOG.info("NApp enabled: %s", napp_id) |
||
131 | except FileExistsError: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Loading history...
|
|||
132 | pass # OK, NApp was already enabled |
||
133 | except PermissionError: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
134 | LOG.error("Failed to enable NApp %s. Permission denied.", |
||
135 | napp_id) |
||
136 | |||
137 | return True |
||
138 | |||
139 | 1 | def disable(self, username, napp_name): |
|
140 | """Disable a NApp if it is enabled.""" |
||
141 | napp_id = "{}/{}".format(username, napp_name) |
||
142 | enabled = self._enabled_path / napp_id |
||
143 | |||
144 | new_manager = NewNAppManager(self._installed_path) |
||
145 | napp = new_manager.napps[napp_id] |
||
146 | deps = napp.napp_dependencies |
||
147 | |||
148 | if deps and napp.meta: |
||
149 | LOG.info('Disabling Meta-NApp %s dependencies: %s', napp, deps) |
||
150 | for uri in deps: |
||
151 | username, napp_name = self.get_napp_fullname_from_uri(uri) |
||
152 | self.disable(username, napp_name) |
||
153 | |||
154 | try: |
||
155 | enabled.unlink() |
||
156 | LOG.info("NApp disabled: %s", napp_id) |
||
157 | if self._controller is not None: |
||
158 | self._controller.unload_napp(username, napp_name) |
||
159 | except FileNotFoundError: |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
160 | pass # OK, it was already disabled |
||
161 | |||
162 | return True |
||
163 | |||
164 | 1 | def enable_all(self): |
|
165 | """Enable all napps already installed and disabled.""" |
||
166 | for napp in self.get_disabled_napps(): |
||
167 | self.enable(napp.username, napp.name) |
||
168 | |||
169 | 1 | def disable_all(self): |
|
170 | """Disable all napps already installed and enabled.""" |
||
171 | for napp in self.get_enabled_napps(): |
||
172 | self.disable(napp.username, napp.name) |
||
173 | |||
174 | 1 | def is_enabled(self, username, napp_name): |
|
175 | """Whether a NApp is enabled or not on this controller FS.""" |
||
176 | napp_id = "{}/{}".format(username, napp_name) |
||
177 | |||
178 | napp = NApp.create_from_uri(napp_id) |
||
179 | return napp in self.get_enabled_napps() |
||
180 | |||
181 | 1 | def is_installed(self, username, napp_name): |
|
182 | """Whether a NApp is installed or not on this controller.""" |
||
183 | napp_id = "{}/{}".format(username, napp_name) |
||
184 | napp = NApp.create_from_uri(napp_id) |
||
185 | return napp in self.get_all_napps() |
||
186 | |||
187 | 1 | @staticmethod |
|
188 | def get_napp_fullname_from_uri(uri): |
||
189 | """Parse URI and get (username, napp_name) tuple.""" |
||
190 | regex = r'^(((https?://|file://)(.+))/)?(.+?)/(.+?)/?(:(.+))?$' |
||
191 | match = re.match(regex, uri) |
||
192 | username = match.groups()[4] |
||
193 | napp_name = match.groups()[5] |
||
194 | return username, napp_name |
||
195 | |||
196 | 1 | def get_all_napps(self): |
|
197 | """List all NApps on this controller FS.""" |
||
198 | return self.get_installed_napps() |
||
199 | |||
200 | 1 | def get_enabled_napps(self): |
|
201 | """Return all enabled NApps on this controller FS.""" |
||
202 | enabled = self.get_napps_from_path(self._enabled_path) |
||
203 | for napp in enabled: |
||
204 | # We should also check if the NApp is enabled on controller |
||
205 | napp.enabled = True |
||
206 | return enabled |
||
207 | |||
208 | 1 | def get_disabled_napps(self): |
|
209 | """Return all disabled NApps on this controller FS.""" |
||
210 | installed = set(self.get_installed_napps()) |
||
211 | enabled = set(self.get_enabled_napps()) |
||
212 | return list(installed - enabled) |
||
213 | |||
214 | 1 | def get_installed_napps(self): |
|
215 | """Return all NApps installed on this controller FS.""" |
||
216 | return self.get_napps_from_path(self._installed_path) |
||
217 | |||
218 | 1 | def get_napp_metadata(self, username, napp_name, key): |
|
219 | """Return a value from kytos.json. |
||
220 | |||
221 | Args: |
||
222 | username (string): A Username. |
||
223 | napp_name (string): A NApp name |
||
224 | key (string): Key used to get the value within kytos.json. |
||
225 | |||
226 | Returns: |
||
227 | meta (object): Value stored in kytos.json. |
||
228 | |||
229 | """ |
||
230 | napp_id = "{}/{}".format(username, napp_name) |
||
231 | kytos_json = self._installed_path / napp_id / 'kytos.json' |
||
232 | try: |
||
233 | with kytos_json.open() as file_descriptor: |
||
234 | meta = json.load(file_descriptor) |
||
235 | return meta[key] |
||
236 | except (FileNotFoundError, json.JSONDecodeError, KeyError): |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
237 | LOG.warning("NApp metadata load failed: %s/kytos.json", napp_id) |
||
238 | return '' |
||
239 | |||
240 | 1 | @staticmethod |
|
241 | 1 | def get_napps_from_path(path: Path): |
|
242 | """List all NApps found in ``napps_dir``.""" |
||
243 | if not path.exists(): |
||
244 | LOG.warning("NApps dir (%s) doesn't exist.", path) |
||
245 | return [] |
||
246 | |||
247 | jsons = path.glob('*/*/kytos.json') |
||
248 | return [NApp.create_from_json(j) for j in jsons] |
||
249 | |||
250 | 1 | @staticmethod |
|
251 | 1 | def _create_module(path: Path): |
|
252 | """Create module with empty __init__.py in `path` if it doesn't exist. |
||
253 | |||
254 | Args: |
||
255 | path: Module path. |
||
256 | """ |
||
257 | if not path.exists(): |
||
258 | path.mkdir(parents=True, exist_ok=True, mode=0o755) |
||
259 | (path / '__init__.py').touch() |
||
260 | |||
261 | 1 | @staticmethod |
|
262 | 1 | def _find_napp(napp, root: Path = None) -> Path: |
|
263 | """Return local NApp root folder. |
||
264 | |||
265 | Search for kytos.json in _./_ folder and _./user/napp_. |
||
266 | |||
267 | Args: |
||
268 | root: Where to begin searching. |
||
269 | |||
270 | Raises: |
||
271 | FileNotFoundError: If there is no such local NApp. |
||
272 | |||
273 | Returns: |
||
274 | NApp root folder. |
||
275 | |||
276 | """ |
||
277 | if root is None: |
||
278 | root = Path() |
||
279 | for folders in ['.'], [napp.username, napp.name]: |
||
280 | kytos_json = root / Path(*folders) / 'kytos.json' |
||
281 | if kytos_json.exists(): |
||
282 | with kytos_json.open() as file_descriptor: |
||
283 | meta = json.load(file_descriptor) |
||
284 | if meta['username'] == napp.username and \ |
||
285 | meta['name'] == napp.name: |
||
286 | return kytos_json.parent |
||
287 | raise FileNotFoundError('kytos.json not found.') |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
288 | |||
289 | |||
290 | 1 | class NewNAppManager: |
|
291 | """A more simple NApp Manager, for just one NApp at a time.""" |
||
292 | |||
293 | 1 | def __init__(self, base_path: Path): |
|
294 | """Create a manager from a NApp base path.""" |
||
295 | self.base_path = base_path |
||
296 | self.napps = {napp.id: napp for napp in self._find_napps()} |
||
297 | |||
298 | 1 | def _find_napps(self): |
|
299 | return NAppsManager.get_napps_from_path(self.base_path) |
||
300 |