diff --git a/README.md b/README.md index d421f43..4298614 100644 --- a/README.md +++ b/README.md @@ -11,15 +11,11 @@ Le tout accessible depuis un seul serveur, avec navigation entre les 3 apps. ## Prérequis - [mise](https://mise.jdx.dev/) — lanceur de tâches -- [uv](https://docs.astral.sh/uv/) — gestion des dépendances et de Python -`uv` installe automatiquement **Python 3.13** et les dépendances (`ldap3`) au premier lancement — rien d'autre à installer. - -Installer mise et uv : +Installer mise : ```bash curl https://mise.run | sh # mise -curl -LsSf https://astral.sh/uv/install.sh | sh # uv ``` ## Installation @@ -27,8 +23,8 @@ curl -LsSf https://astral.sh/uv/install.sh | sh # uv Le dossier `cmdb` est un sous-module git — cloner avec `--recurse-submodules` : ```bash -git clone --recurse-submodules -cd VERSION2 +git clone --recurse-submodules https://git.thehackdes.com/TheHackdes/Dashboard_Admin_Sys.git +cd Dashboard_Admin_Sys ``` Sous-module déjà cloné sans l'option : diff --git a/cert-generator/cert_server.py b/cert-generator/cert_server.py index fae33f9..a7c9d53 100644 --- a/cert-generator/cert_server.py +++ b/cert-generator/cert_server.py @@ -12,68 +12,77 @@ Usage : Sert la RACINE du projet (parent), donc les 3 apps + le switcher fonctionnent. """ + import sys, os, json, base64, shutil, re from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8000 -SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) # .../cert-generator -ROOT = os.path.dirname(SCRIPT_DIR) # racine projet -WORK = os.path.join(SCRIPT_DIR, 'LISTS_ALL_CERT') # dossier de travail (disque) -ARCH = os.path.join(WORK, 'archives') +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) # .../cert-generator +ROOT = os.path.dirname(SCRIPT_DIR) # racine projet +WORK = os.path.join(SCRIPT_DIR, "LISTS_ALL_CERT") # dossier de travail (disque) +ARCH = os.path.join(WORK, "archives") + +SAFE = re.compile(r"^[A-Za-z0-9._@\- ]+$") # noms de dossier/fichier autorisés -SAFE = re.compile(r'^[A-Za-z0-9._@\- ]+$') # noms de dossier/fichier autorisés def safe_name(n): - n = (n or '').strip() - if not n or n in ('.', '..') or '/' in n or '\\' in n or not SAFE.match(n): - raise ValueError('nom invalide: %r' % n) + n = (n or "").strip() + if not n or n in (".", "..") or "/" in n or "\\" in n or not SAFE.match(n): + raise ValueError("nom invalide: %r" % n) return n + def ensure_work(): os.makedirs(WORK, exist_ok=True) + class Handler(SimpleHTTPRequestHandler): def _json(self, obj, code=200): - body = json.dumps(obj).encode('utf-8') + body = json.dumps(obj).encode("utf-8") self.send_response(code) - self.send_header('Content-Type', 'application/json') - self.send_header('Content-Length', str(len(body))) - self.send_header('Cache-Control', 'no-store') + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _err(self, msg, code=400): - self._json({'ok': False, 'error': str(msg)}, code) + self._json({"ok": False, "error": str(msg)}, code) def _body(self): - n = int(self.headers.get('Content-Length', 0)) - return json.loads(self.rfile.read(n) or b'{}') + n = int(self.headers.get("Content-Length", 0)) + return json.loads(self.rfile.read(n) or b"{}") def _api(self): - p = self.path.split('?', 1)[0] - return p.split('/api/', 1)[1] if '/api/' in p else None + p = self.path.split("?", 1)[0] + return p.split("/api/", 1)[1] if "/api/" in p else None # ---------- GET : API list/ping, sinon fichiers statiques ---------- def do_GET(self): api = self._api() - if api == 'ping': - return self._json({'ok': True, 'workdir': 'LISTS_ALL_CERT'}) - if api == 'list': + if api == "ping": + return self._json({"ok": True, "workdir": "LISTS_ALL_CERT"}) + if api == "list": return self._json(self.list_all()) return super().do_GET() def do_POST(self): api = self._api() try: - if api == 'write': return self.write_text() - if api == 'write_bin': return self.write_bin() - if api == 'delete': return self.delete_dir() - if api == 'delete_file': return self.delete_file() - if api == 'delete_archives': return self.delete_archives() + if api == "write": + return self.write_text() + if api == "write_bin": + return self.write_bin() + if api == "delete": + return self.delete_dir() + if api == "delete_file": + return self.delete_file() + if api == "delete_archives": + return self.delete_archives() except Exception as e: return self._err(e, 500) - self._err('route inconnue: %s' % self.path, 404) + self._err("route inconnue: %s" % self.path, 404) # ---------- implémentations ---------- def list_all(self): @@ -81,14 +90,19 @@ class Handler(SimpleHTTPRequestHandler): certs, archives = [], [] for entry in sorted(os.listdir(WORK)): full = os.path.join(WORK, entry) - if entry == 'archives': + if entry == "archives": if os.path.isdir(full): for a in sorted(os.listdir(full)): af = os.path.join(full, a) if os.path.isfile(af): st = os.stat(af) - archives.append({'name': a, 'size': st.st_size, - 'mtime': int(st.st_mtime * 1000)}) + archives.append( + { + "name": a, + "size": st.st_size, + "mtime": int(st.st_mtime * 1000), + } + ) continue if not os.path.isdir(full): continue @@ -100,75 +114,84 @@ class Handler(SimpleHTTPRequestHandler): st = os.stat(ff) m = int(st.st_mtime * 1000) date = max(date, m) - files.append({'name': fn, 'ext': (fn.rsplit('.', 1)[-1] if '.' in fn else '').lower(), - 'size': st.st_size, 'mtime': m}) + files.append( + { + "name": fn, + "ext": (fn.rsplit(".", 1)[-1] if "." in fn else "").lower(), + "size": st.st_size, + "mtime": m, + } + ) if files: - certs.append({'dir': entry, 'files': files, 'date': date}) - archives.sort(key=lambda a: -a['mtime']) - certs.sort(key=lambda c: -c['date']) - return {'ok': True, 'certs': certs, 'archives': archives} + certs.append({"dir": entry, "files": files, "date": date}) + archives.sort(key=lambda a: -a["mtime"]) + certs.sort(key=lambda c: -c["date"]) + return {"ok": True, "certs": certs, "archives": archives} def write_text(self): d = self._body() - dname = safe_name(d['dir']) + dname = safe_name(d["dir"]) target = os.path.join(WORK, dname) os.makedirs(target, exist_ok=True) - for fname, content in (d.get('files') or {}).items(): + for fname, content in (d.get("files") or {}).items(): fn = safe_name(fname) - with open(os.path.join(target, fn), 'w', encoding='utf-8', newline='') as f: + with open(os.path.join(target, fn), "w", encoding="utf-8", newline="") as f: f.write(content) - self._json({'ok': True, 'path': 'LISTS_ALL_CERT/%s/' % dname}) + self._json({"ok": True, "path": "LISTS_ALL_CERT/%s/" % dname}) def write_bin(self): d = self._body() # path = "archives/.zip" OU "/" - parts = [safe_name(x) for x in d['path'].split('/') if x != ''] + parts = [safe_name(x) for x in d["path"].split("/") if x != ""] if not parts: - raise ValueError('path vide') + raise ValueError("path vide") target = os.path.join(WORK, *parts) os.makedirs(os.path.dirname(target), exist_ok=True) - with open(target, 'wb') as f: - f.write(base64.b64decode(d['b64'])) - self._json({'ok': True, 'path': 'LISTS_ALL_CERT/%s' % '/'.join(parts)}) + with open(target, "wb") as f: + f.write(base64.b64decode(d["b64"])) + self._json({"ok": True, "path": "LISTS_ALL_CERT/%s" % "/".join(parts)}) def delete_dir(self): d = self._body() - dname = safe_name(d['dir']) + dname = safe_name(d["dir"]) target = os.path.join(WORK, dname) if os.path.isdir(target): shutil.rmtree(target) - if d.get('archives'): + if d.get("archives"): self._rm_archives(dname) - self._json({'ok': True}) + self._json({"ok": True}) def delete_file(self): d = self._body() - target = os.path.join(WORK, safe_name(d['dir']), safe_name(d['name'])) + target = os.path.join(WORK, safe_name(d["dir"]), safe_name(d["name"])) if os.path.isfile(target): os.remove(target) - self._json({'ok': True}) + self._json({"ok": True}) def delete_archives(self): d = self._body() - self._rm_archives(safe_name(d['dir'])) - self._json({'ok': True}) + self._rm_archives(safe_name(d["dir"])) + self._json({"ok": True}) def _rm_archives(self, dname): low = dname.lower() if os.path.isdir(ARCH): for a in os.listdir(ARCH): if low in a.lower(): - try: os.remove(os.path.join(ARCH, a)) - except OSError: pass + try: + os.remove(os.path.join(ARCH, a)) + except OSError: + pass def log_message(self, *a): pass # silencieux -if __name__ == '__main__': + +if __name__ == "__main__": ensure_work() os.chdir(ROOT) - url = 'http://127.0.0.1:%d/cert-generator/dashboard.html' % PORT - print('Générateur Certificats : %s' % url) - print('Dossier disque : %s (créé automatiquement)' % WORK) - print('Ctrl+C pour arrêter.') - ThreadingHTTPServer(('0.0.0.0', PORT), Handler).serve_forever() + url = "http://127.0.0.1:%d/cert-generator/dashboard.html" % PORT + print("Générateur Certificats : %s" % url) + print("Dossier disque : %s (créé automatiquement)" % WORK) + print("Ctrl+C pour arrêter.") + ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever() diff --git a/cmdb b/cmdb index b06252d..18751d7 160000 --- a/cmdb +++ b/cmdb @@ -1 +1 @@ -Subproject commit b06252daad728064a212411a684e2e95a1a5006e +Subproject commit 18751d75c88a32790a50c1f0a458a2657db6a2ad diff --git a/mise.toml b/mise.toml index 8886c64..9355c26 100644 --- a/mise.toml +++ b/mise.toml @@ -1,5 +1,22 @@ +# Portail Infra — mise config +# https://mise.jdx.dev + +[tools] +python = "3.13" +uv = "latest" + +[env] +# Port par défaut du serveur (surchargé par l'argument CLI de server.py) +PORT = "8000" + [tasks.run] -description = "Run server" -run = ''' -uv run server.py -''' +description = "Lancer le serveur du portail (CMDB + Certificats + Switch)" +run = "uv run server.py {{arg(name='port', default='')}}" + +[tasks.dev] +description = "Lancer le serveur en watch (redémarre à chaque modif .py)" +run = "uv run --with watchfiles watchfiles 'python server.py' server.py cmdb" + +[tasks.clean] +description = "Nettoyer les caches Python" +run = "find . -type d -name __pycache__ -prune -exec rm -rf {} + ; rm -rf .ruff_cache" diff --git a/server.py b/server.py index 2fdddcc..b191af4 100644 --- a/server.py +++ b/server.py @@ -14,30 +14,34 @@ Un seul serveur, lancé à la racine du projet : - Sert les 3 applications + le switcher (liens relatifs OK) - Recherche AD/LDAP de la CMDB conservée (nécessite `pip install ldap3`) """ + import sys, os, json, base64, shutil, re, threading from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs -PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8000 +_arg = sys.argv[1].strip() if len(sys.argv) > 1 else "" +PORT = int(_arg) if _arg else int(os.environ.get("PORT", 8000)) -HERE = os.path.dirname(os.path.abspath(__file__)) # racine projet -CMDB_DIR = os.path.join(HERE, 'cmdb') -CERT_DIR = os.path.join(HERE, 'cert-generator') -WORK = os.path.join(CERT_DIR, 'LISTS_ALL_CERT') # dossier de travail certs (disque) -ARCH = os.path.join(WORK, 'archives') +HERE = os.path.dirname(os.path.abspath(__file__)) # racine projet +CMDB_DIR = os.path.join(HERE, "cmdb") +CERT_DIR = os.path.join(HERE, "cert-generator") +WORK = os.path.join(CERT_DIR, "LISTS_ALL_CERT") # dossier de travail certs (disque) +ARCH = os.path.join(WORK, "archives") # réutilise la logique CMDB existante (base partagée cmdb.db + LDAP) sys.path.insert(0, CMDB_DIR) -import cmdb_server as C # ne démarre rien (guarded __main__) +import cmdb_server as C # ne démarre rien (guarded __main__) + +SAFE = re.compile(r"^[A-Za-z0-9._@\- ]+$") -SAFE = re.compile(r'^[A-Za-z0-9._@\- ]+$') def safe_name(n): - n = (n or '').strip() - if not n or n in ('.', '..') or '/' in n or '\\' in n or not SAFE.match(n): - raise ValueError('nom invalide: %r' % n) + n = (n or "").strip() + if not n or n in (".", "..") or "/" in n or "\\" in n or not SAFE.match(n): + raise ValueError("nom invalide: %r" % n) return n + def ensure_work(): os.makedirs(WORK, exist_ok=True) @@ -48,130 +52,145 @@ class Handler(SimpleHTTPRequestHandler): # ---------- helpers ---------- def _json(self, obj, code=200): - body = json.dumps(obj).encode('utf-8') + body = json.dumps(obj).encode("utf-8") self.send_response(code) - self.send_header('Content-Type', 'application/json') - self.send_header('Content-Length', str(len(body))) - self.send_header('Cache-Control', 'no-store') + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _err(self, msg, code=400): - self._json({'ok': False, 'error': str(msg)}, code) + self._json({"ok": False, "error": str(msg)}, code) def _send_db(self, code, data, version): self.send_response(code) - self.send_header('Content-Type', 'application/octet-stream') - self.send_header('X-DB-Version', str(version)) - self.send_header('Cache-Control', 'no-store') - self.send_header('Content-Length', str(len(data))) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("X-DB-Version", str(version)) + self.send_header("Cache-Control", "no-store") + self.send_header("Content-Length", str(len(data))) self.end_headers() if data: self.wfile.write(data) def _send_download(self, data, filename): self.send_response(200) - self.send_header('Content-Type', 'application/octet-stream') - self.send_header('Content-Disposition', 'attachment; filename="%s"' % filename) - self.send_header('Cache-Control', 'no-store') - self.send_header('Content-Length', str(len(data))) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Content-Disposition", 'attachment; filename="%s"' % filename) + self.send_header("Cache-Control", "no-store") + self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _body(self): - n = int(self.headers.get('Content-Length', 0)) - return json.loads(self.rfile.read(n) or b'{}') + n = int(self.headers.get("Content-Length", 0)) + return json.loads(self.rfile.read(n) or b"{}") def _cert_api(self): - p = self.path.split('?', 1)[0] - if '/cert-generator/api/' in p: - return p.split('/api/', 1)[1] + p = self.path.split("?", 1)[0] + if "/cert-generator/api/" in p: + return p.split("/api/", 1)[1] return None # ---------- GET ---------- def do_GET(self): # racine → CMDB - if self.path in ('/', '/index.html'): + if self.path in ("/", "/index.html"): self.send_response(302) - self.send_header('Location', '/cmdb/cmdb.html') + self.send_header("Location", "/cmdb/cmdb.html") self.end_headers() return # base CMDB partagée - if self.path == '/db': + if self.path == "/db": with C.DB_LOCK: return self._send_db(200, C.db_bytes(), C.db_version()) - if self.path == '/db/version': - return self._json({'version': C.db_version()}) - if self.path == '/db/snapshots': - return self._json({'snapshots': C.list_snapshots()}) - if self.path.startswith('/db/snapshot?'): - name = (parse_qs(urlparse(self.path).query).get('name') or [''])[0] + if self.path == "/db/version": + return self._json({"version": C.db_version()}) + if self.path == "/db/snapshots": + return self._json({"snapshots": C.list_snapshots()}) + if self.path.startswith("/db/snapshot?"): + name = (parse_qs(urlparse(self.path).query).get("name") or [""])[0] try: data = C.snapshot_bytes(name) except (ValueError, OSError): - return self._json({'error': 'snapshot introuvable'}, 404) + return self._json({"error": "snapshot introuvable"}, 404) return self._send_download(data, os.path.basename(name)) # API certs (lecture) api = self._cert_api() - if api == 'ping': - return self._json({'ok': True, 'workdir': 'LISTS_ALL_CERT'}) - if api == 'list': + if api == "ping": + return self._json({"ok": True, "workdir": "LISTS_ALL_CERT"}) + if api == "list": return self._json(self.cert_list()) return super().do_GET() # ---------- POST ---------- def do_POST(self): # base CMDB - if self.path == '/db': - data = self.rfile.read(int(self.headers.get('Content-Length', 0))) - base = int(self.headers.get('X-Base-Version', '0') or 0) + if self.path == "/db": + data = self.rfile.read(int(self.headers.get("Content-Length", 0))) + base = int(self.headers.get("X-Base-Version", "0") or 0) cur = C.db_version() if base != cur: return self._send_db(409, C.db_bytes(), cur) - return self._send_db(200, b'', C.db_write_and_snapshot(data)) + return self._send_db(200, b"", C.db_write_and_snapshot(data)) # snapshots CMDB (création / suppression / restauration) - if self.path in ('/db/snapshot', '/db/snapshot/delete', '/db/restore'): + if self.path in ("/db/snapshot", "/db/snapshot/delete", "/db/restore"): body = self._body() - if self.path == '/db/snapshot': - meta = C.create_snapshot(label=body.get('label') or 'manual') + if self.path == "/db/snapshot": + meta = C.create_snapshot(label=body.get("label") or "manual") if not meta: - return self._json({'error': 'base vide — rien à sauvegarder'}, 400) + return self._json({"error": "base vide — rien à sauvegarder"}, 400) return self._json(meta) - if self.path == '/db/snapshot/delete': + if self.path == "/db/snapshot/delete": try: - C.delete_snapshot(body.get('name') or '') + C.delete_snapshot(body.get("name") or "") except (ValueError, OSError): - return self._json({'error': 'nom de snapshot invalide'}, 400) - return self._json({'ok': True}) + return self._json({"error": "nom de snapshot invalide"}, 400) + return self._json({"ok": True}) try: - C.restore_snapshot(body.get('name') or '') + C.restore_snapshot(body.get("name") or "") except (ValueError, FileNotFoundError): - return self._json({'error': 'snapshot introuvable'}, 404) + return self._json({"error": "snapshot introuvable"}, 404) with C.DB_LOCK: return self._send_db(200, C.db_bytes(), C.db_version()) # LDAP (délégué à cmdb_server) - if self.path in ('/ldap/search', '/ldap/computer'): + if self.path in ("/ldap/search", "/ldap/computer"): if not C.HAS_LDAP: - return self._json({'error': "module ldap3 manquant — pip install ldap3"}, 500) + return self._json( + {"error": "module ldap3 manquant — pip install ldap3"}, 500 + ) try: - body = json.loads(self.rfile.read(int(self.headers.get('Content-Length', 0)))) - fn = C.ldap_computer if self.path == '/ldap/computer' else C.ldap_search - if self.path == '/ldap/search': - return self._json(fn(body.get('config') or {}, (body.get('query') or '').strip(), - int(body.get('limit') or 25))) - return self._json(fn(body.get('config') or {}, (body.get('query') or '').strip())) + body = json.loads( + self.rfile.read(int(self.headers.get("Content-Length", 0))) + ) + fn = C.ldap_computer if self.path == "/ldap/computer" else C.ldap_search + if self.path == "/ldap/search": + return self._json( + fn( + body.get("config") or {}, + (body.get("query") or "").strip(), + int(body.get("limit") or 25), + ) + ) + return self._json( + fn(body.get("config") or {}, (body.get("query") or "").strip()) + ) except Exception as ex: - return self._json({'error': str(ex)}, 500) + return self._json({"error": str(ex)}, 500) # API certs (écriture) api = self._cert_api() try: - if api == 'write': return self.cert_write_text() - if api == 'write_bin': return self.cert_write_bin() - if api == 'delete': return self.cert_delete_dir() - if api == 'delete_archives': return self.cert_delete_archives() + if api == "write": + return self.cert_write_text() + if api == "write_bin": + return self.cert_write_bin() + if api == "delete": + return self.cert_delete_dir() + if api == "delete_archives": + return self.cert_delete_archives() except Exception as e: return self._err(e, 500) - self._err('route inconnue: %s' % self.path, 404) + self._err("route inconnue: %s" % self.path, 404) # ---------- certificats (disque réel) ---------- def cert_list(self): @@ -179,14 +198,19 @@ class Handler(SimpleHTTPRequestHandler): certs, archives = [], [] for entry in sorted(os.listdir(WORK)): full = os.path.join(WORK, entry) - if entry == 'archives': + if entry == "archives": if os.path.isdir(full): for a in sorted(os.listdir(full)): af = os.path.join(full, a) if os.path.isfile(af): st = os.stat(af) - archives.append({'name': a, 'size': st.st_size, - 'mtime': int(st.st_mtime * 1000)}) + archives.append( + { + "name": a, + "size": st.st_size, + "mtime": int(st.st_mtime * 1000), + } + ) continue if not os.path.isdir(full): continue @@ -198,65 +222,78 @@ class Handler(SimpleHTTPRequestHandler): st = os.stat(ff) m = int(st.st_mtime * 1000) date = max(date, m) - files.append({'name': fn, 'ext': (fn.rsplit('.', 1)[-1] if '.' in fn else '').lower(), - 'size': st.st_size, 'mtime': m}) + files.append( + { + "name": fn, + "ext": (fn.rsplit(".", 1)[-1] if "." in fn else "").lower(), + "size": st.st_size, + "mtime": m, + } + ) if files: - certs.append({'dir': entry, 'files': files, 'date': date}) - archives.sort(key=lambda a: -a['mtime']) - certs.sort(key=lambda c: -c['date']) - return {'ok': True, 'certs': certs, 'archives': archives} + certs.append({"dir": entry, "files": files, "date": date}) + archives.sort(key=lambda a: -a["mtime"]) + certs.sort(key=lambda c: -c["date"]) + return {"ok": True, "certs": certs, "archives": archives} def cert_write_text(self): d = self._body() - dname = safe_name(d['dir']) + dname = safe_name(d["dir"]) target = os.path.join(WORK, dname) os.makedirs(target, exist_ok=True) - for fname, content in (d.get('files') or {}).items(): - with open(os.path.join(target, safe_name(fname)), 'w', encoding='utf-8', newline='') as f: + for fname, content in (d.get("files") or {}).items(): + with open( + os.path.join(target, safe_name(fname)), + "w", + encoding="utf-8", + newline="", + ) as f: f.write(content) - self._json({'ok': True, 'path': 'LISTS_ALL_CERT/%s/' % dname}) + self._json({"ok": True, "path": "LISTS_ALL_CERT/%s/" % dname}) def cert_write_bin(self): d = self._body() - parts = [safe_name(x) for x in d['path'].split('/') if x != ''] + parts = [safe_name(x) for x in d["path"].split("/") if x != ""] if not parts: - raise ValueError('path vide') + raise ValueError("path vide") target = os.path.join(WORK, *parts) os.makedirs(os.path.dirname(target), exist_ok=True) - with open(target, 'wb') as f: - f.write(base64.b64decode(d['b64'])) - self._json({'ok': True, 'path': 'LISTS_ALL_CERT/%s' % '/'.join(parts)}) + with open(target, "wb") as f: + f.write(base64.b64decode(d["b64"])) + self._json({"ok": True, "path": "LISTS_ALL_CERT/%s" % "/".join(parts)}) def cert_delete_dir(self): d = self._body() - dname = safe_name(d['dir']) + dname = safe_name(d["dir"]) target = os.path.join(WORK, dname) if os.path.isdir(target): shutil.rmtree(target) - if d.get('archives'): + if d.get("archives"): self._rm_archives(dname) - self._json({'ok': True}) + self._json({"ok": True}) def cert_delete_archives(self): - self._rm_archives(safe_name(self._body()['dir'])) - self._json({'ok': True}) + self._rm_archives(safe_name(self._body()["dir"])) + self._json({"ok": True}) def _rm_archives(self, dname): low = dname.lower() if os.path.isdir(ARCH): for a in os.listdir(ARCH): if low in a.lower(): - try: os.remove(os.path.join(ARCH, a)) - except OSError: pass + try: + os.remove(os.path.join(ARCH, a)) + except OSError: + pass -if __name__ == '__main__': +if __name__ == "__main__": ensure_work() os.chdir(HERE) if not C.HAS_LDAP: print("⚠ ldap3 absent : recherche AD de la CMDB désactivée (pip install ldap3)") - print('Portail Infra : http://127.0.0.1:%d → CMDB' % PORT) - print('CMDB base : %s' % C.DB_PATH) - print('Certificats : %s (créé automatiquement)' % WORK) - print('Ctrl+C pour arrêter.') - ThreadingHTTPServer(('0.0.0.0', PORT), Handler).serve_forever() + print("Portail Infra : http://127.0.0.1:%d → CMDB" % PORT) + print("CMDB base : %s" % C.DB_PATH) + print("Certificats : %s (créé automatiquement)" % WORK) + print("Ctrl+C pour arrêter.") + ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()