#!/usr/bin/env python3 """ Serveur unifié du portail Infra (CMDB + Certificats + Switch). Un seul serveur, lancé à la racine du projet : python3 server.py # http://127.0.0.1:8000 → CMDB python3 server.py 8080 # port custom - http://127.0.0.1:8000/ → redirige vers la CMDB - CMDB : base partagée cmdb.db sourcée automatiquement (/db, versionnée) - Générateur de certificats : écrit sur le disque dans cert-generator/LISTS_ALL_CERT/ (créé au besoin), aucun prompt - 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 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__) 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) return n def ensure_work(): os.makedirs(WORK, exist_ok=True) class Handler(SimpleHTTPRequestHandler): def log_message(self, *a): pass # ---------- helpers ---------- def _json(self, obj, code=200): 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.end_headers() self.wfile.write(body) def _err(self, msg, code=400): 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.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.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'{}') def _cert_api(self): 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'): self.send_response(302) self.send_header('Location', '/cmdb/cmdb.html') self.end_headers() return # base CMDB partagée 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] try: data = C.snapshot_bytes(name) except (ValueError, OSError): 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': 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) 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)) # snapshots CMDB (création / suppression / restauration) 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 not meta: return self._json({'error': 'base vide — rien à sauvegarder'}, 400) return self._json(meta) if self.path == '/db/snapshot/delete': try: C.delete_snapshot(body.get('name') or '') except (ValueError, OSError): return self._json({'error': 'nom de snapshot invalide'}, 400) return self._json({'ok': True}) try: C.restore_snapshot(body.get('name') or '') except (ValueError, FileNotFoundError): 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 not C.HAS_LDAP: 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())) except Exception as ex: 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() except Exception as e: return self._err(e, 500) self._err('route inconnue: %s' % self.path, 404) # ---------- certificats (disque réel) ---------- def cert_list(self): ensure_work() certs, archives = [], [] for entry in sorted(os.listdir(WORK)): full = os.path.join(WORK, entry) 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)}) continue if not os.path.isdir(full): continue files, date = [], 0 for fn in sorted(os.listdir(full)): ff = os.path.join(full, fn) if not os.path.isfile(ff): continue 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}) 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} def cert_write_text(self): d = self._body() 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: f.write(content) 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 != ''] if not parts: 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)}) def cert_delete_dir(self): d = self._body() dname = safe_name(d['dir']) target = os.path.join(WORK, dname) if os.path.isdir(target): shutil.rmtree(target) if d.get('archives'): self._rm_archives(dname) self._json({'ok': True}) def cert_delete_archives(self): 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 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()