Le blog technique

Toutes les astuces #tech des collaborateurs de PI Services.

#openblogPI

Retrouvez les articles à la une

Script Python pour la generation d’un certificat Client Serveur aupres d’une PKI Microsoft

Un certificat qui porte à la fois les usages Server Authentication et Client Authentication est indispensable dès qu’on fait du TLS mutuel : serveurs hors domaine, reverse proxies, API entre services, brokers de messages. Le demander à la main via la console MMC ou certreq reste faisable, mais c’est lent, sujet aux erreurs (SAN oublié, mauvais modèle) et impossible à industrialiser.

Un petit script Python permet de rendre l’opération répétable : une commande, et l’on obtient la clé, le certificat signé par l’autorité Microsoft (AD CS) et un PFX prêt à importer.

#!/usr/bin/env python3
"""
adcs_cert_request.py
--------------------
Génère une clé privée + une CSR (EKU Client Authentication + Server Authentication),
la soumet à une autorité de certification Microsoft (AD CS), récupère le certificat
émis et produit un fichier PFX prêt à importer.

Deux modes de soumission :
  * certreq : utilise certreq.exe (Windows, machine du domaine, auth Kerberos implicite)
  * web     : utilise l'interface Web Enrollment (/certsrv) en NTLM
              -> fonctionne aussi depuis une machine hors domaine (workgroup)

Dépendances :
  pip install cryptography requests requests-ntlm   # requests* seulement pour le mode web

Exemples :
  # Machine du domaine, via certreq
  python adcs_cert_request.py --cn srv01.corp.local --san srv01 --san srv01.corp.local \
      --template ClientServerAuth --method certreq --ca-config "ca01.corp.local\\CORP-CA"

  # Machine en workgroup, via Web Enrollment
  python adcs_cert_request.py --cn gw01.dmz.local --template ClientServerAuth \
      --method web --ca-host ca01.corp.local --user "CORP\\svc_enroll"
"""

import argparse
import getpass
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path

from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import pkcs12
from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID


# --------------------------------------------------------------------------- #
# 1. Clé + CSR
# --------------------------------------------------------------------------- #
def build_csr(cn, sans, key_size, org=None, ou=None, country=None):
    key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)

    name_attrs = [x509.NameAttribute(NameOID.COMMON_NAME, cn)]
    if ou:
        name_attrs.append(x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, ou))
    if org:
        name_attrs.append(x509.NameAttribute(NameOID.ORGANIZATION_NAME, org))
    if country:
        name_attrs.append(x509.NameAttribute(NameOID.COUNTRY_NAME, country))

    # Le CN est toujours repris dans le SAN (les clients TLS modernes ignorent le CN)
    dns_names = list(dict.fromkeys([cn] + (sans or [])))

    csr = (
        x509.CertificateSigningRequestBuilder()
        .subject_name(x509.Name(name_attrs))
        .add_extension(
            x509.SubjectAlternativeName([x509.DNSName(n) for n in dns_names]),
            critical=False,
        )
        .add_extension(
            x509.ExtendedKeyUsage(
                [ExtendedKeyUsageOID.SERVER_AUTH, ExtendedKeyUsageOID.CLIENT_AUTH]
            ),
            critical=False,
        )
        .add_extension(
            x509.KeyUsage(
                digital_signature=True, key_encipherment=True,
                content_commitment=False, data_encipherment=False,
                key_agreement=False, key_cert_sign=False, crl_sign=False,
                encipher_only=False, decipher_only=False,
            ),
            critical=True,
        )
        .sign(key, hashes.SHA256())
    )
    return key, csr


# --------------------------------------------------------------------------- #
# 2a. Soumission via certreq.exe
# --------------------------------------------------------------------------- #
def submit_certreq(csr_path, template, ca_config, out_dir):
    if os.name != "nt":
        sys.exit("Le mode certreq nécessite Windows (certreq.exe).")
    cer_path = out_dir / (csr_path.stem + ".cer")
    cmd = ["certreq.exe", "-submit", "-q",
           "-attrib", f"CertificateTemplate:{template}"]
    if ca_config:
        cmd += ["-config", ca_config]
    cmd += [str(csr_path), str(cer_path)]

    print(f"[*] {' '.join(cmd)}")
    res = subprocess.run(cmd, capture_output=True, text=True)
    print(res.stdout.strip())
    if res.returncode != 0 or not cer_path.exists():
        req_id = re.search(r"RequestId:\s*\"?(\d+)", res.stdout)
        if req_id:
            sys.exit(f"[!] Requête en attente ou refusée (RequestId {req_id.group(1)}). "
                     f"Récupération ultérieure : certreq -retrieve {req_id.group(1)} out.cer")
        sys.exit(f"[!] Échec certreq : {res.stderr.strip()}")
    return x509.load_pem_x509_certificate(cer_path.read_bytes()) \
        if b"BEGIN CERTIFICATE" in cer_path.read_bytes() \
        else x509.load_der_x509_certificate(cer_path.read_bytes())


# --------------------------------------------------------------------------- #
# 2b. Soumission via Web Enrollment (/certsrv)
# --------------------------------------------------------------------------- #
def submit_web(csr_pem, template, ca_host, user, password, verify):
    import requests
    from requests_ntlm import HttpNtlmAuth

    base = f"https://{ca_host}/certsrv"
    s = requests.Session()
    s.auth = HttpNtlmAuth(user, password)
    s.verify = verify

    data = {
        "Mode": "newreq",
        "CertRequest": csr_pem,
        "CertAttrib": f"CertificateTemplate:{template}",
        "TargetStoreFlags": "0",
        "SaveCert": "yes",
    }
    print(f"[*] POST {base}/certfnsh.asp (modèle {template})")
    r = s.post(f"{base}/certfnsh.asp", data=data, timeout=60)
    if r.status_code == 401:
        sys.exit("[!] Authentification refusée (vérifier compte / NTLM activé sur IIS).")
    r.raise_for_status()

    m = re.search(r"certnew\.cer\?ReqID=(\d+)", r.text)
    if not m:
        pending = re.search(r"Your Request Id is (\d+)", r.text)
        if pending:
            sys.exit(f"[!] Requête en attente d'approbation (ReqID {pending.group(1)}). "
                     f"Récupérer plus tard : {base}/certnew.cer?ReqID={pending.group(1)}&Enc=b64")
        err = re.search(r"The disposition message is \"([^\"]+)", r.text)
        sys.exit(f"[!] Requête refusée : {err.group(1) if err else 'réponse inattendue de /certsrv'}")

    req_id = m.group(1)
    print(f"[+] Certificat émis (ReqID {req_id}), téléchargement…")
    c = s.get(f"{base}/certnew.cer", params={"ReqID": req_id, "Enc": "b64"}, timeout=60)
    c.raise_for_status()
    return x509.load_pem_x509_certificate(c.content)


# --------------------------------------------------------------------------- #
# 3. Main
# --------------------------------------------------------------------------- #
def main():
    p = argparse.ArgumentParser(description="CSR client/serveur + soumission AD CS")
    p.add_argument("--cn", required=True, help="Common Name (FQDN de la machine)")
    p.add_argument("--san", action="append", default=[], help="SAN DNS (répétable)")
    p.add_argument("--org"); p.add_argument("--ou"); p.add_argument("--country")
    p.add_argument("--key-size", type=int, default=2048, choices=[2048, 3072, 4096])
    p.add_argument("--template", required=True, help="Nom (pas le display name) du modèle AD CS")
    p.add_argument("--method", choices=["certreq", "web"], default="certreq")
    p.add_argument("--ca-config", help='certreq : "serveurCA\\NomCA" (sinon fenêtre de choix)')
    p.add_argument("--ca-host", help="web : FQDN du serveur Web Enrollment")
    p.add_argument("--user", help='web : compte "DOMAINE\\user"')
    p.add_argument("--ca-bundle", help="web : fichier PEM de la CA racine pour valider le TLS")
    p.add_argument("--insecure", action="store_true", help="web : ne pas valider le TLS (test)")
    p.add_argument("--out-dir", default=".", help="Répertoire de sortie")
    p.add_argument("--no-pfx", action="store_true", help="Ne pas générer de PFX")
    a = p.parse_args()

    out = Path(a.out_dir); out.mkdir(parents=True, exist_ok=True)
    base = a.cn.replace("*", "wildcard")

    # Clé + CSR
    key, csr = build_csr(a.cn, a.san, a.key_size, a.org, a.ou, a.country)
    pfx_pwd = None if a.no_pfx else getpass.getpass("Mot de passe du PFX (et de la clé) : ")
    enc = (serialization.BestAvailableEncryption(pfx_pwd.encode())
           if pfx_pwd else serialization.NoEncryption())

    key_path = out / f"{base}.key"
    key_path.write_bytes(key.private_bytes(serialization.Encoding.PEM,
                                           serialization.PrivateFormat.PKCS8, enc))
    try:
        os.chmod(key_path, 0o600)
    except OSError:
        pass
    csr_pem = csr.public_bytes(serialization.Encoding.PEM).decode()
    csr_path = out / f"{base}.csr"
    csr_path.write_text(csr_pem)
    print(f"[+] Clé : {key_path}\n[+] CSR : {csr_path}")

    # Soumission
    if a.method == "certreq":
        cert = submit_certreq(csr_path, a.template, a.ca_config, out)
    else:
        if not (a.ca_host and a.user):
            sys.exit("--ca-host et --user sont requis en mode web.")
        pwd = os.environ.get("ADCS_PASSWORD") or getpass.getpass(f"Mot de passe de {a.user} : ")
        verify = False if a.insecure else (a.ca_bundle or True)
        cert = submit_web(csr_pem, a.template, a.ca_host, a.user, pwd, verify)

    cer_path = out / f"{base}.cer"
    cer_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
    print(f"[+] Certificat : {cer_path}")
    print(f"    Sujet  : {cert.subject.rfc4514_string()}")
    print(f"    Émis par : {cert.issuer.rfc4514_string()}")
    print(f"    Valide jusqu'au : {cert.not_valid_after_utc:%Y-%m-%d}")

    # Contrôle des EKU réellement émis (le modèle de la CA a le dernier mot)
    try:
        ekus = cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value
        for oid, label in [(ExtendedKeyUsageOID.SERVER_AUTH, "Server Auth"),
                           (ExtendedKeyUsageOID.CLIENT_AUTH, "Client Auth")]:
            print(f"    {label:12}: {'OK' if oid in ekus else 'ABSENT !'}")
    except x509.ExtensionNotFound:
        print("    [!] Aucun EKU dans le certificat émis.")

    # PFX
    if not a.no_pfx:
        pfx = pkcs12.serialize_key_and_certificates(
            name=a.cn.encode(), key=key, cert=cert, cas=None,
            encryption_algorithm=serialization.BestAvailableEncryption(pfx_pwd.encode()),
        )
        pfx_path = out / f"{base}.pfx"
        pfx_path.write_bytes(pfx)
        print(f"[+] PFX : {pfx_path}")
        print(f'    Import (Windows, admin) : certutil -f -p "<mdp>" -importpfx "{pfx_path}"')


if __name__ == "__main__":
    main()

Ce que fait le script

adcs_cert_request.py enchaîne quatre étapes, avec la seule bibliothèque cryptography pour la partie PKI.

  1. Clé et CSR : génération d’une clé RSA (2048 bits par défaut) et d’une demande signée en SHA-256, avec le CN, les SAN DNS, les EKU Server Auth + Client Auth et le Key Usage adapté.
  2. Soumission à la CA, au choix :
    • certreq : appelle certreq.exe sur une machine Windows du domaine, l’authentification Kerberos est implicite ;
    • web : poste la demande sur l’interface Web Enrollment (/certsrv) en NTLM, ce qui fonctionne aussi depuis une machine en workgroup ou en DMZ.
  3. Contrôle : le script vérifie que les deux EKU figurent bien dans le certificat émis, car c’est le modèle côté CA qui a le dernier mot.
  4. Export PFX : clé et certificat sont regroupés dans un PKCS#12 protégé par mot de passe, importable avec certutil -importpfx.

Si la demande reste en attente d’approbation d’un gestionnaire de certificats, le script affiche le RequestId et la commande pour récupérer le certificat plus tard.

Prérequis côté CA

Tout se joue dans le modèle de certificat ; sans lui, la meilleure CSR du monde sera refusée.

  • Modèle dédié : dupliquer Web Server (ou Computer), ajouter Client Authentication dans les stratégies d’application, et cocher Fournir dans la demande dans l’onglet Nom du sujet pour que CN et SAN soient repris.
  • Nom technique : --template attend le nom du modèle, pas son nom complet affiché (souvent le même sans espaces).
  • Permissions : le compte qui soumet a besoin des droits Lire et Inscrire sur le modèle, et le modèle doit être publié sur la CA.
  • Mode web : le rôle Web Enrollment doit être installé, publié en HTTPS, avec l’authentification Windows (NTLM) active sur IIS.

Utilisation

Installation des dépendances (requests et requests-ntlm ne servent qu’au mode web) :

pip install cryptography requests requests-ntlm

Sur un serveur membre du domaine, via certreq :

python adcs_cert_request.py --cn srv01.corp.local --san srv01 `
    --template ClientServerAuth --method certreq --ca-config "ca01.corp.local\CORP-CA"

Sur une machine en workgroup, via Web Enrollment :

python adcs_cert_request.py --cn gw01.dmz.local --template ClientServerAuth \
    --method web --ca-host ca01.corp.local --user "CORP\svc_enroll" --ca-bundle root-ca.pem

Le script produit quatre fichiers : .key, .csr, .cer et .pfx. Le mot de passe du compte peut aussi venir de la variable d’environnement ADCS_PASSWORD, pratique en pipeline.

[Le Saviez-vous ?] – Retrouver la complexité du Pin Bitlocker

Saviez vous qu’il était possible de retrouver la complexité, la longueur et d’autres paramètres du Pin Bitlocker via la base de registre ?

Récemment je me suis retrouvé sur poste chiffré avec Bitlocker et un Pin au démarrage mais, je n’avais pas le Pin…
Forte heureusement avec la clé de récupération il m’a été possible de démarrer et d’ouvrir une session, puis avec l’aide de Powershell (avec d’un compte admin) et les commandes manage-bde appliquer un nouveau Pin.
Mais alors que je me pensais sorti d’affaire, le Pin que j’ai voulu appliqué ne respectait pas la politique (que je n’avais pas non plus… Décidément), c’est finalement grâce à la base de registre que j’ai pu obtenir ces informations.

Si vous aussi avez besoin de connaitre la politique Bitlocker, il suffit de vérifier ici :

HKEY_LOCAL_MACHINE\SOFTWARE\Policies\Microsoft\FVE

La longueur minimale du Pin se trouve sous l’entrée « MinimumPIN« 

La valeur qui définit si le Pin contient seulement des chiffres où, des majuscules, minuscules, chiffres et caractères spéciaux (Enhanced Pin) se trouve sous l’entrée « UseEnhancedPin » (0 est égale à désactivé, 1 est égale à activé)

Gérer la concurrence dans vos scripts Bash avec flock

Lorsque plusieurs instances d’un même script Bash s’exécutent en parallèle, des problèmes de concurrence (race conditions) peuvent vite apparaître : écriture simultanée dans un fichier journal, exécution en double d’une tâche d’arrière-plan ou saturation des ressources système.

Pour éviter qu’un script ne s’exécute deux fois en même temps (notamment dans une tâche cron), la commande flock est la solution idéale sous Linux.

Comment fonctionne flock ?

flock repose sur des verrous au niveau du système de fichiers (file locks). Tant qu’un processus détient le verrou sur un fichier spécifique, aucun autre processus ne peut obtenir ce même verrou.

1. Empêcher l’exécution simultanée d’un script

Le cas d’usage le plus courant consiste à utiliser flock pour encapsuler l’appel d’un script.

Bash

flock -n /tmp/mon_script.lock /chemin/vers/mon_script.sh

  • -n (non-blocking) : si le verrou est déjà pris par une autre instance, flock s’arrête immédiatement sans exécuter la commande.
  • /tmp/mon_script.lock : le fichier qui sert de témoin de verrouillage.

Gestion des verrous à l’intérieur d’un script

Il est également possible d’intégrer flock directement dans le script lui-même pour qu’il gère son propre verrouillage dès le démarrage.

Bash

#!/usr/bin/env bash

exec 200>/var/run/mon_script.lock
flock -n 200 || { echo "Une autre instance est déjà en cours d'exécution." ; exit 1 ; }

# Votre logique de script ici
echo "Traitement en cours..."
sleep 10

Explication du fonctionnement :

  1. exec 200>/var/run/mon_script.lock ouvre le fichier sur le descripteur de fichier 200.
  2. flock -n 200 essaie d’acquérir le verrou sur ce descripteur.
  3. Si l’acquisition échoue, le script affiche un message et s’arrête proprement.
  4. À la fin du script (ou en cas de crash), le descripteur est fermé et le verrou est automatiquement libéré par le noyau.

Les options indispensables

OptionDescription
-n (--nonblock)Échoue immédiatement si le verrou n’est pas disponible (par défaut, flock attend qu’il se libère).
-u (--unlock)Libère explicitement le verrou (rarement nécessaire car géré à la fermeture du processus).
-w <secondes>Attend un temps défini pour obtenir le verrou avant d’abandonner.
-s (--shared)Pose un verrou partagé (lecture seule autorisée pour plusieurs processus).
-x (--exclusive)Pose un verrou exclusif (comportement par défaut).

Pourquoi préférer flock à un simple fichier .pid ?

La création manuelle d’un fichier temporaire (ex. touch /tmp/script.lock) pose problème en cas de plantage ou de coupure de courant : le fichier reste sur le disque et bloque les exécutions futures.

flock résout ce problème car le verrou est géré par le noyau Linux et lié au processus en cours. Si le processus meurt, le verrou saute immédiatement.