Initial commit: CMDB dashboard

Serveur Python (http.server + ldap3) servant cmdb.html, la base SQLite
partagee avec versioning optimiste, snapshots horodates et recherche AD.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,40 @@
+# Database (runtime data)
+cmdb.db
+cmdb.db.ver
+*.db
+*.db-journal
+*.db-wal
+*.db-shm
+
+# Snapshots (generated backups)
+snapshots/
+
+# Python
+__pycache__/
+*.py[cod]
+*.egg-info/
+.eggs/
+build/
+dist/
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# uv
+.uv/
+
+# Env / secrets
+.env
+.env.*
+*.secret
+
+# IDE
+.vscode/
+.idea/
+*.swp
+
+# OS
+.DS_Store
+Thumbs.db

README.md

@@ -0,0 +1,76 @@
+<div align="center">
+  <img src="sources/logo.png" alt="Manage OpenLDAP Logo" width="220"/>
+  
+  <h1>Noctâme</h1>
+  <p><i>CMDB légère : inventaire de parc et recherche Active Directory</i></p>
+</div>
+
+---
+
+## À propos
+
+Noctâme est une CMDB minimaliste. Un seul serveur Python sert l'interface
+(`cmdb.html`), partage une base SQLite entre plusieurs utilisateurs et
+interroge l'annuaire Active Directory.
+
+## Démarrage
+
+Prérequis : [mise](https://mise.jdx.dev/) (gère la version de Python) et
+[uv](https://docs.astral.sh/uv/) (gère les dépendances).
+
+```bash
+mise install          # installe Python (>=3.13)
+uv sync               # installe les dépendances (ldap3)
+
+uv run cmdb_server.py         # http://0.0.0.0:8000
+uv run cmdb_server.py 8080    # port custom
+```
+
+## Fonctionnalités
+
+- Inventaire du parc dans une base SQLite partagée (`cmdb.db`)
+- Détection de conflit à l'écriture (concurrence optimiste)
+- Snapshots automatiques et manuels de la base
+- Recherche Active Directory / LDAP intégrée
+
+---
+
+## Annexe
+
+<details>
+<summary>Génération du logo</summary>
+
+</br>
+
+1. **Prompt pour ChatGPT (DALL-E)** :
+
+```
+A high-quality, original creature inspired by Pokémon style, unevolved stage (basic form). The creature has balanced proportions (can be cute, neutral, or slightly serious), with a clean and memorable silhouette, expressive eyes (not overly exaggerated), and a distinct personality that can range from friendly to calm, mysterious, or determined.
+
+The creature has a unique elemental theme (choose one: fire, water, electric, grass, ghost, etc.), with subtle and well-integrated visual cues reflecting its type (glowing sparks, leaves, flames, mist, energy patterns, etc.).
+
+Art style: official Pokémon-style illustration, soft but controlled shading, vibrant yet natural color palette, smooth gradients, clean and precise outlines, slightly stylized but grounded anatomy.
+
+Pose: simple, confident, and readable, facing slightly toward the viewer, with a composed or expressive stance depending on its personality.
+
+Background: transparent.
+
+Lighting: soft studio lighting with gentle contrast to enhance form and details.
+
+Composition: centered, full body visible, no cropping.
+
+Aspect ratio: 1:1
+
+High resolution, sharp details, no text, no watermark.
+
+Theme: Astro, Star, Milkyway
+
+Format d'image : **1:1**
+```
+
+2. **Supprimer le fond** : [PhotoRoom Background Remover](https://www.photoroom.com/tools/background-remover)
+
+3. **Size Logo :** 220x200
+
+</details>
+

cmdb_server.py

@@ -0,0 +1,482 @@
+#!/usr/bin/env python3
+"""Serveur CMDB : sert cmdb.html, la base partagée et la recherche AD.
+
+Remplace `python -m http.server` ET le connecteur séparé :
+    pip install ldap3
+    python3 cmdb_server.py            # http://0.0.0.0:8000
+    python3 cmdb_server.py 8080       # port custom
+
+Endpoints :
+    GET  /db          → base SQLite (bytes) + entête X-DB-Version
+    GET  /db/version  → {"version": n}   (poll léger)
+    POST /db          → écrit la base ; entête X-Base-Version pour la
+                        concurrence optimiste (409 si obsolète)
+    POST /ldap/search → recherche annuaire (config envoyée par la page)
+
+La base est un vrai fichier `cmdb.db` partagé : plusieurs personnes
+peuvent l'utiliser en même temps, avec détection de conflit.
+"""
+import json
+import os
+import re
+import ssl
+import sys
+import threading
+import time
+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__))
+DB_PATH = os.path.join(HERE, 'cmdb.db')
+VER_PATH = DB_PATH + '.ver'
+DB_LOCK = threading.Lock()
+
+
+def db_version():
+    try:
+        with open(VER_PATH) as f:
+            return int(f.read().strip() or 0)
+    except (OSError, ValueError):
+        return 1 if os.path.exists(DB_PATH) else 0
+
+
+def db_bytes():
+    try:
+        with open(DB_PATH, 'rb') as f:
+            return f.read()
+    except OSError:
+        return b''
+
+
+def db_write(data):
+    """Écrit la base atomiquement et incrémente la version. Renvoie la version."""
+    with DB_LOCK:
+        ver = db_version() + 1
+        tmp = DB_PATH + '.tmp'
+        with open(tmp, 'wb') as f:
+            f.write(data)
+        os.replace(tmp, DB_PATH)
+        with open(VER_PATH, 'w') as f:
+            f.write(str(ver))
+        return ver
+
+
+# ---------------------------------------------------------------------------
+# Snapshots : copies horodatées de la base dans snapshots/.
+#   - « auto »   : créées après une écriture, au plus 1/heure, rétention 72.
+#   - « manuel » : créées à la demande (bouton), rétention 100, label libre.
+# Restaurer un snapshot crée d'abord un snapshot de sécurité de l'état courant.
+# ---------------------------------------------------------------------------
+SNAP_DIR = os.path.join(HERE, 'snapshots')
+# nom : cmdb-<AAAAMMJJ>-<HHMMSS>-v<version>[-<label>].db  (pas de '/' possible → pas de traversal)
+SNAP_RE = re.compile(r'^cmdb-(\d{8})-(\d{6})-v(\d+)(?:-([A-Za-z0-9_.-]+))?\.db$')
+AUTO_MIN_INTERVAL = 3600   # s : au plus un snapshot auto par heure
+AUTO_KEEP = 72             # rétention snapshots auto (~3 jours horaires)
+MANUAL_KEEP = 100          # rétention snapshots manuels
+
+
+def _sanitize_label(label):
+    label = re.sub(r'[^A-Za-z0-9_.-]+', '-', (label or '').strip()).strip('-')
+    return label[:40]
+
+
+def _parse_snap(name):
+    """Valide un nom de snapshot et renvoie sa méta, ou None si invalide."""
+    name = os.path.basename(name or '')
+    m = SNAP_RE.match(name)
+    if not m:
+        return None
+    return {'name': name, 'date': m.group(1), 'time': m.group(2),
+            'version': int(m.group(3)), 'label': m.group(4) or '',
+            'auto': m.group(4) == 'auto'}
+
+
+def list_snapshots():
+    """Liste les snapshots existants, plus récents d'abord."""
+    out = []
+    try:
+        entries = os.listdir(SNAP_DIR)
+    except OSError:
+        return out
+    for n in entries:
+        meta = _parse_snap(n)
+        if not meta:
+            continue
+        try:
+            st = os.stat(os.path.join(SNAP_DIR, n))
+        except OSError:
+            continue
+        meta['size'] = st.st_size
+        meta['mtime'] = int(st.st_mtime * 1000)
+        out.append(meta)
+    out.sort(key=lambda s: s['mtime'], reverse=True)
+    return out
+
+
+def _prune(auto, keep):
+    for s in [s for s in list_snapshots() if s['auto'] == auto][keep:]:
+        try:
+            os.remove(os.path.join(SNAP_DIR, s['name']))
+        except OSError:
+            pass
+
+
+def create_snapshot(label='manual', auto=False):
+    """Copie la base courante dans snapshots/. Renvoie la méta, ou None si base vide."""
+    with DB_LOCK:
+        data = db_bytes()
+        if not data:
+            return None
+        ver = db_version()
+        os.makedirs(SNAP_DIR, exist_ok=True)
+        lbl = 'auto' if auto else (_sanitize_label(label) or 'manual')
+        name = 'cmdb-%s-v%d-%s.db' % (time.strftime('%Y%m%d-%H%M%S'), ver, lbl)
+        path = os.path.join(SNAP_DIR, name)
+        tmp = path + '.tmp'
+        with open(tmp, 'wb') as f:
+            f.write(data)
+        os.replace(tmp, path)
+    _prune(auto, AUTO_KEEP if auto else MANUAL_KEEP)
+    return _parse_snap(name)
+
+
+def _maybe_auto_snapshot():
+    """Snapshot auto si le dernier date de plus d'AUTO_MIN_INTERVAL. Jamais dans DB_LOCK."""
+    autos = [s for s in list_snapshots() if s['auto']]
+    if autos and (time.time() - autos[0]['mtime'] / 1000) < AUTO_MIN_INTERVAL:
+        return
+    try:
+        create_snapshot(auto=True)
+    except OSError:
+        pass
+
+
+def db_write_and_snapshot(data):
+    """Écrit la base puis déclenche un snapshot auto (throttlé). Renvoie la version."""
+    ver = db_write(data)
+    _maybe_auto_snapshot()
+    return ver
+
+
+def snapshot_bytes(name):
+    """Contenu d'un snapshot. Lève si nom invalide ou fichier absent."""
+    meta = _parse_snap(name)
+    if not meta:
+        raise ValueError('nom de snapshot invalide')
+    with open(os.path.join(SNAP_DIR, meta['name']), 'rb') as f:
+        return f.read()
+
+
+def delete_snapshot(name):
+    meta = _parse_snap(name)
+    if not meta:
+        raise ValueError('nom de snapshot invalide')
+    path = os.path.join(SNAP_DIR, meta['name'])
+    if os.path.isfile(path):
+        os.remove(path)
+
+
+def restore_snapshot(name):
+    """Remplace la base courante par le snapshot `name`.
+
+    Crée d'abord un snapshot de sécurité de l'état courant. Renvoie la nouvelle version.
+    """
+    meta = _parse_snap(name)
+    if not meta:
+        raise ValueError('nom de snapshot invalide')
+    path = os.path.join(SNAP_DIR, meta['name'])
+    if not os.path.isfile(path):
+        raise FileNotFoundError('snapshot introuvable')
+    with open(path, 'rb') as f:
+        data = f.read()
+    create_snapshot(label='pre-restore')   # filet de sécurité avant écrasement
+    return db_write(data)
+
+try:
+    from ldap3 import Server, Connection, Tls, SUBTREE, ALL_ATTRIBUTES
+    from ldap3.utils.conv import escape_filter_chars
+    HAS_LDAP = True
+except ImportError:
+    HAS_LDAP = False
+
+
+def _cn_of(dn):
+    if not dn:
+        return ''
+    head = dn.split(',')[0]
+    return head.split('=', 1)[1] if '=' in head else head
+
+
+def ldap_connect(cfg):
+    """Ouvre et bind une connexion LDAP selon la config de la page."""
+    use_ssl = cfg.get('method') == 'ssl'
+    tls = Tls(validate=ssl.CERT_REQUIRED if cfg.get('verify_certificates') else ssl.CERT_NONE)
+    srv = Server(cfg['host'], port=int(cfg.get('port') or (636 if use_ssl else 389)),
+                 use_ssl=use_ssl, tls=tls, get_info=None, connect_timeout=8)
+    conn = Connection(srv, user=cfg.get('bind_dn') or None,
+                      password=cfg.get('bind_password') or None,
+                      auto_referrals=False, receive_timeout=15)
+    if cfg.get('method') == 'start_tls':
+        conn.open()
+        conn.start_tls()
+    if not conn.bind():
+        raise RuntimeError('Bind LDAP refusé : %s' % conn.result)
+    return conn
+
+
+def ldap_computer(cfg, query):
+    """Cherche un poste (objectClass=computer) par nom/dNSHostName."""
+    conn = ldap_connect(cfg)
+    q = escape_filter_chars((query or '').strip().rstrip('$'))
+    flt = ('(&(objectClass=computer)(|(cn=*%s*)(name=*%s*)(dNSHostName=*%s*)'
+           '(sAMAccountName=*%s*)))' % (q, q, q, q))
+    base = cfg.get('base') or ''
+    ok = conn.search(base, flt, SUBTREE, attributes=ALL_ATTRIBUTES, size_limit=5)
+    res = conn.result or {}
+    if not ok and res.get('result') not in (0, 4):
+        raise RuntimeError('Recherche LDAP échouée : %s' % (res.get('description') or res))
+    out = []
+    for e in conn.entries:
+        d = e.entry_attributes_as_dict
+
+        def val(k):
+            for kk in d:
+                if kk.lower() == k.lower():
+                    v = d[kk]
+                    if isinstance(v, list):
+                        v = v[0] if v else ''
+                    return str(v) if v else ''
+            return ''
+
+        def vals(k):
+            for kk in d:
+                if kk.lower() == k.lower():
+                    v = d[kk]
+                    return [str(x) for x in (v if isinstance(v, list) else [v]) if x]
+            return []
+
+        uac = val('userAccountControl')
+        enabled = (not (int(uac) & 2)) if uac.isdigit() else None
+        out.append({
+            'name': val('cn') or val('name'),
+            'dns': val('dNSHostName'),
+            'os': val('operatingSystem'),
+            'os_version': val('operatingSystemVersion'),
+            'description': val('description'),
+            'location': val('location'),
+            'managed_by': _cn_of(val('managedBy')),
+            'when_created': val('whenCreated')[:8],
+            'enabled': enabled,
+            'groups': [_cn_of(x) for x in vals('memberOf')][:20],
+        })
+    return out
+
+
+def ldap_search(cfg, query, limit=25):
+    """Recherche d'utilisateurs dans l'AD selon la config de la page.
+
+    Cherche par username, nom, prénom, nom affiché ou mail.
+    Lève une exception explicite si le bind ou la recherche échoue
+    (sinon un problème de config renverrait « aucun résultat » en silence).
+    """
+    conn = ldap_connect(cfg)
+
+    uid = cfg.get('uid') or 'userPrincipalName'
+    raw = cfg.get('filter') or ''
+    if raw:
+        flt = raw.replace('{query}', escape_filter_chars(query))
+    else:
+        # recherche prioritairement sur displayName, multi-mots (ET) → "jean dupont"
+        # et "dupont jean" fonctionnent tous les deux, quel que soit l'ordre.
+        search_attrs = ['displayName', 'cn', 'sAMAccountName', 'givenName', 'sn',
+                        'userPrincipalName', 'mail']
+        tokens = [escape_filter_chars(t) for t in query.split() if t] or ['']
+        parts = []
+        for t in tokens:
+            ors = ''.join('(%s=*%s*)' % (a, t) for a in search_attrs)
+            parts.append('(|%s)' % ors)
+        flt = '(&%s)' % ''.join(parts) if len(parts) > 1 else parts[0]
+    # personnes uniquement, compatible AD + OpenLDAP
+    person = ('(|(objectClass=user)(objectClass=inetOrgPerson)'
+              '(objectClass=organizationalPerson)(objectClass=person))')
+    base = cfg.get('base') or ''
+
+    ok = conn.search(base, '(&%s%s)' % (person, flt), SUBTREE,
+                     attributes=ALL_ATTRIBUTES, size_limit=limit)
+    res = conn.result or {}
+    # 0 = success, 4 = sizeLimitExceeded (résultats partiels OK)
+    if not ok and res.get('result') not in (0, 4):
+        raise RuntimeError('Recherche LDAP échouée : %s (base=%s)'
+                           % (res.get('description') or res, base or '∅'))
+
+    def val(d, k):
+        # accès insensible à la casse, première valeur
+        for kk in d:
+            if kk.lower() == k.lower():
+                v = d[kk]
+                if isinstance(v, list):
+                    v = v[0] if v else ''
+                return str(v) if v else ''
+        return ''
+
+    def vals(d, k):
+        for kk in d:
+            if kk.lower() == k.lower():
+                v = d[kk]
+                return [str(x) for x in (v if isinstance(v, list) else [v]) if x]
+        return []
+
+    def cn_of(dn):
+        # "CN=Jean Dupont,OU=..." -> "Jean Dupont"
+        if not dn:
+            return ''
+        head = dn.split(',')[0]
+        return head.split('=', 1)[1] if '=' in head else head
+
+    out, seen = [], set()
+    for e in conn.entries:
+        d = e.entry_attributes_as_dict
+        name = val(d, 'displayName') or val(d, 'cn') \
+            or (val(d, 'givenName') + ' ' + val(d, 'sn')).strip()
+        username = val(d, 'sAMAccountName') or val(d, uid) or val(d, 'userPrincipalName')
+        key = (name, username)
+        if not (name or username) or key in seen:
+            continue
+        seen.add(key)
+        uac = val(d, 'userAccountControl')
+        enabled = (not (int(uac) & 2)) if uac.isdigit() else None
+        out.append({
+            'displayName': name or username,
+            'username': username,
+            'mail': val(d, 'mail'),
+            'title': val(d, 'title'),
+            'department': val(d, 'department'),
+            'company': val(d, 'company'),
+            'office': val(d, 'physicalDeliveryOfficeName'),
+            'phone': val(d, 'telephoneNumber'),
+            'mobile': val(d, 'mobile') or val(d, 'ipPhone'),
+            'manager': cn_of(val(d, 'manager')),
+            'description': val(d, 'description'),
+            'employee_id': val(d, 'employeeID') or val(d, 'employeeNumber'),
+            'division': val(d, 'division'),
+            'city': val(d, 'l'),
+            'country': val(d, 'co') or val(d, 'c'),
+            'enabled': enabled,
+            'groups': [cn_of(x) for x in vals(d, 'memberOf')][:15],
+        })
+    out.sort(key=lambda u: u['displayName'].lower())
+    return out
+
+
+class Handler(SimpleHTTPRequestHandler):
+    def log_message(self, *args):
+        pass
+
+    def _json(self, code, payload):
+        body = json.dumps(payload).encode()
+        self.send_response(code)
+        self.send_header('Content-Type', 'application/json')
+        self.send_header('Content-Length', str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    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 do_GET(self):
+        if self.path == '/db':
+            with DB_LOCK:
+                return self._send_db(200, db_bytes(), db_version())
+        if self.path == '/db/version':
+            return self._json(200, {'version': db_version()})
+        if self.path == '/db/snapshots':
+            return self._json(200, {'snapshots': list_snapshots()})
+        if self.path.startswith('/db/snapshot?'):
+            name = (parse_qs(urlparse(self.path).query).get('name') or [''])[0]
+            try:
+                data = snapshot_bytes(name)
+            except (ValueError, OSError):
+                return self._json(404, {'error': 'snapshot introuvable'})
+            return self._send_download(data, os.path.basename(name))
+        if self.path in ('/', '/index.html'):
+            self.path = '/cmdb.html'
+        return super().do_GET()
+
+    def do_POST(self):
+        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 = db_version()
+            if base != cur:                       # quelqu'un a écrit entre-temps
+                return self._send_db(409, db_bytes(), cur)
+            ver = db_write_and_snapshot(data)
+            return self._send_db(200, b'', ver)
+        if self.path in ('/db/snapshot', '/db/snapshot/delete', '/db/restore'):
+            try:
+                body = json.loads(self.rfile.read(int(self.headers.get('Content-Length', 0)) or 0) or b'{}')
+            except ValueError:
+                body = {}
+            if self.path == '/db/snapshot':
+                meta = create_snapshot(label=body.get('label') or 'manual')
+                if not meta:
+                    return self._json(400, {'error': 'base vide — rien à sauvegarder'})
+                return self._json(200, meta)
+            if self.path == '/db/snapshot/delete':
+                try:
+                    delete_snapshot(body.get('name') or '')
+                except (ValueError, OSError):
+                    return self._json(400, {'error': 'nom de snapshot invalide'})
+                return self._json(200, {'ok': True})
+            # /db/restore
+            try:
+                restore_snapshot(body.get('name') or '')
+            except (ValueError, FileNotFoundError):
+                return self._json(404, {'error': 'snapshot introuvable'})
+            with DB_LOCK:
+                return self._send_db(200, db_bytes(), db_version())
+        if self.path == '/ldap/computer':
+            if not HAS_LDAP:
+                return self._json(500, {'error': "module ldap3 manquant — pip install ldap3"})
+            try:
+                body = json.loads(self.rfile.read(int(self.headers.get('Content-Length', 0))))
+                result = ldap_computer(body.get('config') or {}, (body.get('query') or '').strip())
+                return self._json(200, result)
+            except Exception as ex:
+                return self._json(500, {'error': str(ex)})
+        if self.path != '/ldap/search':
+            return self._json(404, {'error': 'not found'})
+        if not HAS_LDAP:
+            return self._json(500, {'error': "module ldap3 manquant — pip install ldap3"})
+        try:
+            body = json.loads(self.rfile.read(int(self.headers.get('Content-Length', 0))))
+            limit = int(body.get('limit') or 25)
+            result = ldap_search(body.get('config') or {}, (body.get('query') or '').strip(), limit)
+            self._json(200, result)
+        except Exception as ex:  # renvoyé tel quel à la page
+            self._json(500, {'error': str(ex)})
+
+
+if __name__ == '__main__':
+    os.chdir(os.path.dirname(os.path.abspath(__file__)))
+    if not HAS_LDAP:
+        print("⚠ ldap3 absent : la recherche AD renverra une erreur (pip install ldap3)")
+    print('CMDB : http://127.0.0.1:%d  (Ctrl+C pour arrêter)' % PORT)
+    ThreadingHTTPServer(('0.0.0.0', PORT), Handler).serve_forever()

pyproject.toml

@@ -0,0 +1,9 @@
+[project]
+name = "cmdb"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "ldap3>=2.9.1",
+]