Add JWT authentication with user/password login, role-based access control, and Bearer token support while maintaining legacy API key compatibility for transition; update README with comprehensive JWT security guidelines, installation instructions for Python/SSH, development startup scripts (.env, run_dev.sh/ps1), and migrate all API examples to JWT authentication; update Ansible inventory SSH key paths from Docker container paths to local user paths
This commit is contained in:
@@ -0,0 +1,16 @@
|
||||
"""
|
||||
Utilitaires pour l'API Homelab Automation.
|
||||
"""
|
||||
|
||||
from app.utils.ssh_utils import find_ssh_private_key, run_ssh_command, bootstrap_host
|
||||
from app.utils.pdf_generator import markdown_to_pdf_bytes
|
||||
from app.utils.markdown_parser import HelpHtmlToMarkdownParser, build_help_markdown
|
||||
|
||||
__all__ = [
|
||||
"find_ssh_private_key",
|
||||
"run_ssh_command",
|
||||
"bootstrap_host",
|
||||
"markdown_to_pdf_bytes",
|
||||
"HelpHtmlToMarkdownParser",
|
||||
"build_help_markdown",
|
||||
]
|
||||
@@ -0,0 +1,287 @@
|
||||
"""
|
||||
Parser HTML vers Markdown pour la documentation d'aide.
|
||||
"""
|
||||
|
||||
import re
|
||||
from html.parser import HTMLParser
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
|
||||
class HelpHtmlToMarkdownParser(HTMLParser):
|
||||
"""Parser pour convertir le HTML de documentation en Markdown."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.markdown_lines: List[str] = []
|
||||
self.current_text = ""
|
||||
self.in_code = False
|
||||
self.in_pre = False
|
||||
self.in_list = False
|
||||
self.list_type = "ul"
|
||||
self.list_level = 0
|
||||
self.in_table = False
|
||||
self.table_row: List[str] = []
|
||||
self.in_th = False
|
||||
self.ignore_content = False
|
||||
self.tag_stack: List[str] = []
|
||||
|
||||
def handle_starttag(self, tag: str, attrs: List[Tuple[str, str]]):
|
||||
self.tag_stack.append(tag)
|
||||
attrs_dict = dict(attrs)
|
||||
|
||||
if tag in ["script", "style", "nav", "footer"]:
|
||||
self.ignore_content = True
|
||||
return
|
||||
|
||||
if tag == "h1":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n# ")
|
||||
elif tag == "h2":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n## ")
|
||||
elif tag == "h3":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n### ")
|
||||
elif tag == "h4":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n#### ")
|
||||
elif tag == "p":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n")
|
||||
elif tag == "br":
|
||||
self.markdown_lines.append("\n")
|
||||
elif tag == "strong" or tag == "b":
|
||||
self.current_text += "**"
|
||||
elif tag == "em" or tag == "i":
|
||||
self.current_text += "*"
|
||||
elif tag == "code":
|
||||
if not self.in_pre:
|
||||
self.current_text += "`"
|
||||
self.in_code = True
|
||||
elif tag == "pre":
|
||||
self._flush_text()
|
||||
self.in_pre = True
|
||||
self.markdown_lines.append("\n```\n")
|
||||
elif tag == "ul":
|
||||
self._flush_text()
|
||||
self.in_list = True
|
||||
self.list_type = "ul"
|
||||
self.list_level += 1
|
||||
elif tag == "ol":
|
||||
self._flush_text()
|
||||
self.in_list = True
|
||||
self.list_type = "ol"
|
||||
self.list_level += 1
|
||||
elif tag == "li":
|
||||
self._flush_text()
|
||||
indent = " " * (self.list_level - 1)
|
||||
if self.list_type == "ul":
|
||||
self.markdown_lines.append(f"\n{indent}- ")
|
||||
else:
|
||||
self.markdown_lines.append(f"\n{indent}1. ")
|
||||
elif tag == "a":
|
||||
href = attrs_dict.get("href", "")
|
||||
self.current_text += "["
|
||||
self._href_pending = href
|
||||
elif tag == "table":
|
||||
self._flush_text()
|
||||
self.in_table = True
|
||||
self.markdown_lines.append("\n")
|
||||
elif tag == "tr":
|
||||
self.table_row = []
|
||||
elif tag == "th":
|
||||
self.in_th = True
|
||||
elif tag == "td":
|
||||
pass
|
||||
elif tag == "hr":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n---\n")
|
||||
elif tag == "blockquote":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n> ")
|
||||
|
||||
def handle_endtag(self, tag: str):
|
||||
if self.tag_stack and self.tag_stack[-1] == tag:
|
||||
self.tag_stack.pop()
|
||||
|
||||
if tag in ["script", "style", "nav", "footer"]:
|
||||
self.ignore_content = False
|
||||
return
|
||||
|
||||
if tag in ["h1", "h2", "h3", "h4"]:
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n")
|
||||
elif tag == "p":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n")
|
||||
elif tag == "strong" or tag == "b":
|
||||
self.current_text += "**"
|
||||
elif tag == "em" or tag == "i":
|
||||
self.current_text += "*"
|
||||
elif tag == "code":
|
||||
if not self.in_pre:
|
||||
self.current_text += "`"
|
||||
self.in_code = False
|
||||
elif tag == "pre":
|
||||
self._flush_text()
|
||||
self.in_pre = False
|
||||
self.markdown_lines.append("```\n")
|
||||
elif tag == "ul" or tag == "ol":
|
||||
self._flush_text()
|
||||
self.list_level -= 1
|
||||
if self.list_level == 0:
|
||||
self.in_list = False
|
||||
self.markdown_lines.append("\n")
|
||||
elif tag == "li":
|
||||
self._flush_text()
|
||||
elif tag == "a":
|
||||
href = getattr(self, "_href_pending", "")
|
||||
self.current_text += f"]({href})"
|
||||
self._href_pending = ""
|
||||
elif tag == "tr":
|
||||
if self.table_row:
|
||||
self.markdown_lines.append("| " + " | ".join(self.table_row) + " |\n")
|
||||
if self.in_th:
|
||||
# Ajouter la ligne de séparation après les en-têtes
|
||||
self.markdown_lines.append("|" + "|".join(["---"] * len(self.table_row)) + "|\n")
|
||||
self.in_th = False
|
||||
elif tag == "th" or tag == "td":
|
||||
self._flush_text()
|
||||
self.table_row.append(self.current_text.strip())
|
||||
self.current_text = ""
|
||||
elif tag == "table":
|
||||
self.in_table = False
|
||||
self.markdown_lines.append("\n")
|
||||
elif tag == "blockquote":
|
||||
self._flush_text()
|
||||
self.markdown_lines.append("\n")
|
||||
|
||||
def handle_data(self, data: str):
|
||||
if self.ignore_content:
|
||||
return
|
||||
|
||||
if self.in_pre:
|
||||
self.markdown_lines.append(data)
|
||||
else:
|
||||
# Normaliser les espaces
|
||||
text = " ".join(data.split())
|
||||
if text:
|
||||
self.current_text += text
|
||||
|
||||
def _flush_text(self):
|
||||
if self.current_text.strip():
|
||||
self.markdown_lines.append(self.current_text.strip())
|
||||
self.current_text = ""
|
||||
|
||||
def get_markdown(self) -> str:
|
||||
self._flush_text()
|
||||
content = "".join(self.markdown_lines)
|
||||
# Nettoyer les lignes vides multiples
|
||||
content = re.sub(r'\n{3,}', '\n\n', content)
|
||||
return content.strip()
|
||||
|
||||
|
||||
def build_help_markdown(html_path: Path = None, html_content: str = None) -> str:
|
||||
"""Construit le contenu Markdown d'aide depuis un fichier HTML.
|
||||
|
||||
Args:
|
||||
html_path: Chemin vers le fichier HTML source
|
||||
html_content: Contenu HTML direct (prioritaire sur html_path)
|
||||
|
||||
Returns:
|
||||
Contenu Markdown formaté
|
||||
"""
|
||||
if html_content:
|
||||
content = html_content
|
||||
elif html_path and html_path.exists():
|
||||
content = html_path.read_text(encoding='utf-8')
|
||||
else:
|
||||
return _get_default_help_markdown()
|
||||
|
||||
# Extraire uniquement la section d'aide si présente
|
||||
help_section_match = re.search(
|
||||
r'<section[^>]*id=["\']help["\'][^>]*>(.*?)</section>',
|
||||
content,
|
||||
re.DOTALL | re.IGNORECASE
|
||||
)
|
||||
|
||||
if help_section_match:
|
||||
help_html = help_section_match.group(1)
|
||||
else:
|
||||
# Utiliser tout le body si pas de section help
|
||||
body_match = re.search(r'<body[^>]*>(.*?)</body>', content, re.DOTALL | re.IGNORECASE)
|
||||
help_html = body_match.group(1) if body_match else content
|
||||
|
||||
# Parser le HTML
|
||||
parser = HelpHtmlToMarkdownParser()
|
||||
parser.feed(help_html)
|
||||
|
||||
return parser.get_markdown()
|
||||
|
||||
|
||||
def _get_default_help_markdown() -> str:
|
||||
"""Retourne le contenu d'aide par défaut."""
|
||||
return """# 📚 Documentation Homelab Automation
|
||||
|
||||
## 🎯 Introduction
|
||||
|
||||
Bienvenue dans le **Homelab Automation Dashboard**, une plateforme complète pour gérer votre infrastructure homelab avec Ansible.
|
||||
|
||||
## 🖥️ Dashboard
|
||||
|
||||
Le dashboard principal affiche:
|
||||
- **Hôtes en ligne**: Nombre d'hôtes actuellement accessibles
|
||||
- **Tâches**: Nombre total de tâches exécutées
|
||||
- **Taux de succès**: Pourcentage de tâches réussies
|
||||
- **Uptime**: Disponibilité globale du système
|
||||
|
||||
## 🖧 Gestion des Hôtes
|
||||
|
||||
### Ajouter un hôte
|
||||
1. Cliquez sur "Nouvel hôte"
|
||||
2. Renseignez le nom et l'adresse IP
|
||||
3. Sélectionnez le groupe d'environnement
|
||||
4. Choisissez les groupes de rôles
|
||||
5. Cliquez sur "Ajouter"
|
||||
|
||||
### Bootstrap
|
||||
Le bootstrap configure un hôte pour Ansible:
|
||||
- Création de l'utilisateur d'automatisation
|
||||
- Configuration des clés SSH
|
||||
- Installation de sudo et Python3
|
||||
|
||||
## 📋 Playbooks
|
||||
|
||||
Les playbooks Ansible sont organisés par catégorie:
|
||||
- **Général**: Playbooks de base
|
||||
- **Maintenance**: Mises à jour, redémarrages
|
||||
- **Sécurité**: Configurations de sécurité
|
||||
- **Monitoring**: Surveillance et alertes
|
||||
|
||||
## ⏰ Schedules
|
||||
|
||||
Planifiez l'exécution automatique de playbooks:
|
||||
- **Une fois**: Exécution unique à une date/heure
|
||||
- **Quotidien**: Tous les jours à une heure fixe
|
||||
- **Hebdomadaire**: Certains jours de la semaine
|
||||
- **Mensuel**: Un jour spécifique du mois
|
||||
- **Personnalisé**: Expression cron
|
||||
|
||||
## 🔔 Notifications
|
||||
|
||||
Les notifications sont envoyées via ntfy:
|
||||
- Démarrage/arrêt de l'application
|
||||
- Succès/échec des tâches
|
||||
- Bootstrap réussi/échoué
|
||||
|
||||
## 🔑 API
|
||||
|
||||
L'API REST est accessible avec une clé API:
|
||||
```
|
||||
Header: X-API-Key: votre-cle-api
|
||||
```
|
||||
|
||||
---
|
||||
*Généré par Homelab Automation Dashboard*
|
||||
"""
|
||||
@@ -0,0 +1,323 @@
|
||||
"""
|
||||
Générateur PDF à partir de contenu Markdown.
|
||||
"""
|
||||
|
||||
import io
|
||||
import re
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from reportlab.lib import colors
|
||||
from reportlab.lib.pagesizes import A4
|
||||
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
|
||||
from reportlab.lib.units import cm
|
||||
from reportlab.platypus import (
|
||||
SimpleDocTemplate,
|
||||
Paragraph,
|
||||
Spacer,
|
||||
Preformatted,
|
||||
Table,
|
||||
TableStyle,
|
||||
PageBreak,
|
||||
)
|
||||
|
||||
|
||||
def extract_leading_emojis(text: str) -> Tuple[str, str]:
|
||||
"""Extrait les emojis en début de texte.
|
||||
|
||||
Args:
|
||||
text: Texte potentiellement commençant par des emojis
|
||||
|
||||
Returns:
|
||||
Tuple (emojis, texte_restant)
|
||||
"""
|
||||
emoji_pattern = re.compile(
|
||||
"["
|
||||
"\U0001F600-\U0001F64F" # emoticons
|
||||
"\U0001F300-\U0001F5FF" # symbols & pictographs
|
||||
"\U0001F680-\U0001F6FF" # transport & map symbols
|
||||
"\U0001F1E0-\U0001F1FF" # flags
|
||||
"\U00002702-\U000027B0" # dingbats
|
||||
"\U000024C2-\U0001F251" # enclosed characters
|
||||
"\U0001F900-\U0001F9FF" # supplemental symbols
|
||||
"\U0001FA00-\U0001FA6F" # chess symbols
|
||||
"\U0001FA70-\U0001FAFF" # symbols extended
|
||||
"\U00002600-\U000026FF" # misc symbols
|
||||
"]+"
|
||||
)
|
||||
|
||||
match = emoji_pattern.match(text)
|
||||
if match:
|
||||
emojis = match.group()
|
||||
rest = text[len(emojis):].lstrip()
|
||||
return emojis, rest
|
||||
|
||||
return "", text
|
||||
|
||||
|
||||
def emoji_to_png_bytes(emoji: str, size: int = 32) -> Optional[bytes]:
|
||||
"""Convertit un emoji en image PNG.
|
||||
|
||||
Note: Nécessite Pillow avec support de polices emoji.
|
||||
|
||||
Args:
|
||||
emoji: Caractère emoji
|
||||
size: Taille en pixels
|
||||
|
||||
Returns:
|
||||
Bytes de l'image PNG ou None si échec
|
||||
"""
|
||||
try:
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
# Créer une image transparente
|
||||
img = Image.new('RGBA', (size, size), (255, 255, 255, 0))
|
||||
draw = ImageDraw.Draw(img)
|
||||
|
||||
# Essayer différentes polices emoji
|
||||
font_candidates = [
|
||||
"/usr/share/fonts/truetype/noto/NotoColorEmoji.ttf",
|
||||
"/System/Library/Fonts/Apple Color Emoji.ttc",
|
||||
"C:\\Windows\\Fonts\\seguiemj.ttf",
|
||||
"/usr/share/fonts/google-noto-emoji/NotoColorEmoji.ttf",
|
||||
]
|
||||
|
||||
font = None
|
||||
for font_path in font_candidates:
|
||||
try:
|
||||
font = ImageFont.truetype(font_path, size - 4)
|
||||
break
|
||||
except (OSError, IOError):
|
||||
continue
|
||||
|
||||
if font is None:
|
||||
font = ImageFont.load_default()
|
||||
|
||||
# Dessiner l'emoji centré
|
||||
bbox = draw.textbbox((0, 0), emoji, font=font)
|
||||
text_width = bbox[2] - bbox[0]
|
||||
text_height = bbox[3] - bbox[1]
|
||||
x = (size - text_width) // 2
|
||||
y = (size - text_height) // 2
|
||||
draw.text((x, y), emoji, font=font, embedded_color=True)
|
||||
|
||||
# Convertir en bytes
|
||||
buffer = io.BytesIO()
|
||||
img.save(buffer, format='PNG')
|
||||
buffer.seek(0)
|
||||
return buffer.getvalue()
|
||||
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def markdown_to_pdf_bytes(markdown_content: str, title: str = "Document") -> bytes:
|
||||
"""Convertit du contenu Markdown en PDF.
|
||||
|
||||
Args:
|
||||
markdown_content: Contenu au format Markdown
|
||||
title: Titre du document
|
||||
|
||||
Returns:
|
||||
Bytes du PDF généré
|
||||
"""
|
||||
buffer = io.BytesIO()
|
||||
doc = SimpleDocTemplate(
|
||||
buffer,
|
||||
pagesize=A4,
|
||||
rightMargin=2*cm,
|
||||
leftMargin=2*cm,
|
||||
topMargin=2*cm,
|
||||
bottomMargin=2*cm
|
||||
)
|
||||
|
||||
# Styles
|
||||
styles = getSampleStyleSheet()
|
||||
|
||||
title_style = ParagraphStyle(
|
||||
'CustomTitle',
|
||||
parent=styles['Heading1'],
|
||||
fontSize=24,
|
||||
spaceAfter=30,
|
||||
textColor=colors.HexColor('#1a1a2e'),
|
||||
)
|
||||
|
||||
h1_style = ParagraphStyle(
|
||||
'CustomH1',
|
||||
parent=styles['Heading1'],
|
||||
fontSize=18,
|
||||
spaceBefore=20,
|
||||
spaceAfter=12,
|
||||
textColor=colors.HexColor('#16213e'),
|
||||
)
|
||||
|
||||
h2_style = ParagraphStyle(
|
||||
'CustomH2',
|
||||
parent=styles['Heading2'],
|
||||
fontSize=14,
|
||||
spaceBefore=15,
|
||||
spaceAfter=8,
|
||||
textColor=colors.HexColor('#0f3460'),
|
||||
)
|
||||
|
||||
h3_style = ParagraphStyle(
|
||||
'CustomH3',
|
||||
parent=styles['Heading3'],
|
||||
fontSize=12,
|
||||
spaceBefore=10,
|
||||
spaceAfter=6,
|
||||
textColor=colors.HexColor('#1a1a2e'),
|
||||
)
|
||||
|
||||
body_style = ParagraphStyle(
|
||||
'CustomBody',
|
||||
parent=styles['Normal'],
|
||||
fontSize=10,
|
||||
spaceBefore=4,
|
||||
spaceAfter=4,
|
||||
leading=14,
|
||||
)
|
||||
|
||||
code_style = ParagraphStyle(
|
||||
'CustomCode',
|
||||
parent=styles['Code'],
|
||||
fontSize=8,
|
||||
fontName='Courier',
|
||||
backColor=colors.HexColor('#f4f4f4'),
|
||||
borderColor=colors.HexColor('#e0e0e0'),
|
||||
borderWidth=1,
|
||||
borderPadding=5,
|
||||
spaceBefore=8,
|
||||
spaceAfter=8,
|
||||
)
|
||||
|
||||
bullet_style = ParagraphStyle(
|
||||
'CustomBullet',
|
||||
parent=body_style,
|
||||
leftIndent=20,
|
||||
bulletIndent=10,
|
||||
)
|
||||
|
||||
# Parser le Markdown
|
||||
elements = []
|
||||
elements.append(Paragraph(title, title_style))
|
||||
elements.append(Spacer(1, 20))
|
||||
|
||||
lines = markdown_content.split('\n')
|
||||
in_code_block = False
|
||||
code_block_content = []
|
||||
in_table = False
|
||||
table_rows = []
|
||||
|
||||
for line in lines:
|
||||
# Blocs de code
|
||||
if line.strip().startswith('```'):
|
||||
if in_code_block:
|
||||
# Fin du bloc de code
|
||||
code_text = '\n'.join(code_block_content)
|
||||
if code_text.strip():
|
||||
elements.append(Preformatted(code_text, code_style))
|
||||
code_block_content = []
|
||||
in_code_block = False
|
||||
else:
|
||||
# Début du bloc de code
|
||||
in_code_block = True
|
||||
continue
|
||||
|
||||
if in_code_block:
|
||||
code_block_content.append(line)
|
||||
continue
|
||||
|
||||
# Tables Markdown
|
||||
if '|' in line and not line.strip().startswith('#'):
|
||||
cells = [c.strip() for c in line.split('|')]
|
||||
cells = [c for c in cells if c] # Supprimer les cellules vides
|
||||
|
||||
if cells and not all(c.replace('-', '').replace(':', '') == '' for c in cells):
|
||||
if not in_table:
|
||||
in_table = True
|
||||
table_rows.append(cells)
|
||||
continue
|
||||
elif in_table and table_rows:
|
||||
# Fin de la table
|
||||
if len(table_rows) > 1:
|
||||
# Créer la table
|
||||
table = Table(table_rows)
|
||||
table.setStyle(TableStyle([
|
||||
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#e0e0e0')),
|
||||
('TEXTCOLOR', (0, 0), (-1, 0), colors.HexColor('#1a1a2e')),
|
||||
('ALIGN', (0, 0), (-1, -1), 'LEFT'),
|
||||
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
|
||||
('FONTSIZE', (0, 0), (-1, -1), 9),
|
||||
('BOTTOMPADDING', (0, 0), (-1, 0), 8),
|
||||
('TOPPADDING', (0, 0), (-1, -1), 4),
|
||||
('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#cccccc')),
|
||||
]))
|
||||
elements.append(table)
|
||||
elements.append(Spacer(1, 10))
|
||||
table_rows = []
|
||||
in_table = False
|
||||
|
||||
# Titres
|
||||
if line.startswith('# '):
|
||||
emojis, text = extract_leading_emojis(line[2:].strip())
|
||||
display_text = f"{emojis} {text}".strip() if emojis else text
|
||||
elements.append(Paragraph(display_text, h1_style))
|
||||
elif line.startswith('## '):
|
||||
emojis, text = extract_leading_emojis(line[3:].strip())
|
||||
display_text = f"{emojis} {text}".strip() if emojis else text
|
||||
elements.append(Paragraph(display_text, h2_style))
|
||||
elif line.startswith('### '):
|
||||
emojis, text = extract_leading_emojis(line[4:].strip())
|
||||
display_text = f"{emojis} {text}".strip() if emojis else text
|
||||
elements.append(Paragraph(display_text, h3_style))
|
||||
elif line.startswith('#### '):
|
||||
text = line[5:].strip()
|
||||
elements.append(Paragraph(f"<b>{text}</b>", body_style))
|
||||
|
||||
# Listes à puces
|
||||
elif line.strip().startswith('- ') or line.strip().startswith('* '):
|
||||
text = line.strip()[2:]
|
||||
# Convertir le markdown basique
|
||||
text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
|
||||
text = re.sub(r'`(.+?)`', r'<font face="Courier">\1</font>', text)
|
||||
elements.append(Paragraph(f"• {text}", bullet_style))
|
||||
|
||||
# Listes numérotées
|
||||
elif re.match(r'^\d+\.\s', line.strip()):
|
||||
match = re.match(r'^(\d+)\.\s(.+)', line.strip())
|
||||
if match:
|
||||
num, text = match.groups()
|
||||
text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
|
||||
text = re.sub(r'`(.+?)`', r'<font face="Courier">\1</font>', text)
|
||||
elements.append(Paragraph(f"{num}. {text}", bullet_style))
|
||||
|
||||
# Ligne horizontale
|
||||
elif line.strip() in ['---', '***', '___']:
|
||||
elements.append(Spacer(1, 10))
|
||||
|
||||
# Paragraphe normal
|
||||
elif line.strip():
|
||||
text = line.strip()
|
||||
# Convertir le markdown basique
|
||||
text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
|
||||
text = re.sub(r'\*(.+?)\*', r'<i>\1</i>', text)
|
||||
text = re.sub(r'`(.+?)`', r'<font face="Courier" color="#c7254e">\1</font>', text)
|
||||
elements.append(Paragraph(text, body_style))
|
||||
|
||||
else:
|
||||
# Ligne vide
|
||||
elements.append(Spacer(1, 6))
|
||||
|
||||
# Traiter la dernière table si présente
|
||||
if in_table and table_rows and len(table_rows) > 1:
|
||||
table = Table(table_rows)
|
||||
table.setStyle(TableStyle([
|
||||
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#e0e0e0')),
|
||||
('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#cccccc')),
|
||||
]))
|
||||
elements.append(table)
|
||||
|
||||
# Générer le PDF
|
||||
doc.build(elements)
|
||||
buffer.seek(0)
|
||||
return buffer.getvalue()
|
||||
@@ -0,0 +1,318 @@
|
||||
"""
|
||||
Utilitaires SSH pour le bootstrap et les connexions distantes.
|
||||
"""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from app.core.config import settings
|
||||
from app.schemas.common import CommandResult
|
||||
|
||||
|
||||
def find_ssh_private_key() -> Optional[str]:
|
||||
"""Trouve une clé SSH privée valide parmi les emplacements candidats.
|
||||
|
||||
Recherche dans l'ordre:
|
||||
1. Chemin configuré (SSH_KEY_PATH)
|
||||
2. Clés Docker (/app/docker/ssh_keys/)
|
||||
3. Clés utilisateur (~/.ssh/)
|
||||
|
||||
Returns:
|
||||
Chemin vers la clé privée ou None si aucune trouvée
|
||||
"""
|
||||
# Candidats dans l'ordre de priorité
|
||||
candidates = []
|
||||
|
||||
# 1. Chemin configuré
|
||||
if settings.ssh_key_path:
|
||||
candidates.append(Path(settings.ssh_key_path))
|
||||
|
||||
# 2. Clés Docker (pour environnement conteneurisé)
|
||||
docker_ssh_dir = Path("/app/docker/ssh_keys")
|
||||
if docker_ssh_dir.exists():
|
||||
for key_name in ["id_automation_ansible", "id_rsa", "id_ed25519", "id_ecdsa"]:
|
||||
candidates.append(docker_ssh_dir / key_name)
|
||||
|
||||
# 3. Clés utilisateur standard
|
||||
user_ssh_dir = Path.home() / ".ssh"
|
||||
if user_ssh_dir.exists():
|
||||
for key_name in ["id_rsa", "id_ed25519", "id_ecdsa", "id_dsa"]:
|
||||
candidates.append(user_ssh_dir / key_name)
|
||||
|
||||
# Retourner la première clé existante
|
||||
for candidate in candidates:
|
||||
if candidate.exists() and candidate.is_file():
|
||||
return str(candidate)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def find_ssh_public_key() -> Optional[str]:
|
||||
"""Trouve une clé SSH publique valide.
|
||||
|
||||
Returns:
|
||||
Contenu de la clé publique ou None si aucune trouvée
|
||||
"""
|
||||
private_key = find_ssh_private_key()
|
||||
if not private_key:
|
||||
return None
|
||||
|
||||
public_key_path = Path(private_key + ".pub")
|
||||
if public_key_path.exists():
|
||||
return public_key_path.read_text().strip()
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def run_ssh_command(
|
||||
host: str,
|
||||
command: str,
|
||||
user: str = "root",
|
||||
password: str = None,
|
||||
private_key: str = None,
|
||||
timeout: int = 30
|
||||
) -> CommandResult:
|
||||
"""Exécute une commande SSH sur un hôte distant.
|
||||
|
||||
Args:
|
||||
host: Adresse IP ou hostname de l'hôte
|
||||
command: Commande à exécuter
|
||||
user: Utilisateur SSH (défaut: root)
|
||||
password: Mot de passe (si authentification par mot de passe)
|
||||
private_key: Chemin vers la clé privée (optionnel)
|
||||
timeout: Timeout en secondes
|
||||
|
||||
Returns:
|
||||
CommandResult avec le résultat de la commande
|
||||
"""
|
||||
if password:
|
||||
# Vérifier que sshpass est disponible
|
||||
if not shutil.which("sshpass"):
|
||||
return CommandResult(
|
||||
status="error",
|
||||
return_code=-1,
|
||||
stdout="",
|
||||
stderr="sshpass non installé. Installez-le avec: apt-get install sshpass"
|
||||
)
|
||||
|
||||
ssh_cmd = [
|
||||
"sshpass", "-p", password,
|
||||
"ssh",
|
||||
"-o", "StrictHostKeyChecking=no",
|
||||
"-o", "UserKnownHostsFile=/dev/null",
|
||||
"-o", f"ConnectTimeout={timeout}",
|
||||
f"{user}@{host}",
|
||||
command
|
||||
]
|
||||
else:
|
||||
# Authentification par clé
|
||||
if not private_key:
|
||||
private_key = find_ssh_private_key()
|
||||
|
||||
ssh_cmd = [
|
||||
"ssh",
|
||||
"-o", "StrictHostKeyChecking=no",
|
||||
"-o", "UserKnownHostsFile=/dev/null",
|
||||
"-o", "BatchMode=yes",
|
||||
"-o", f"ConnectTimeout={timeout}",
|
||||
]
|
||||
|
||||
if private_key:
|
||||
ssh_cmd.extend(["-i", private_key])
|
||||
|
||||
ssh_cmd.extend([f"{user}@{host}", command])
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
ssh_cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout + 10
|
||||
)
|
||||
|
||||
return CommandResult(
|
||||
status="success" if result.returncode == 0 else "error",
|
||||
return_code=result.returncode,
|
||||
stdout=result.stdout,
|
||||
stderr=result.stderr
|
||||
)
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return CommandResult(
|
||||
status="timeout",
|
||||
return_code=-1,
|
||||
stdout="",
|
||||
stderr=f"Timeout après {timeout} secondes"
|
||||
)
|
||||
except Exception as e:
|
||||
return CommandResult(
|
||||
status="error",
|
||||
return_code=-1,
|
||||
stdout="",
|
||||
stderr=str(e)
|
||||
)
|
||||
|
||||
|
||||
def bootstrap_host(
|
||||
host: str,
|
||||
root_password: str,
|
||||
automation_user: str = None
|
||||
) -> CommandResult:
|
||||
"""Bootstrap un hôte pour Ansible.
|
||||
|
||||
Cette fonction:
|
||||
1. Se connecte via SSH avec le mot de passe root
|
||||
2. Crée l'utilisateur d'automatisation
|
||||
3. Configure la clé SSH publique
|
||||
4. Configure sudo sans mot de passe
|
||||
5. Installe Python3 si nécessaire
|
||||
6. Vérifie la connexion SSH par clé
|
||||
|
||||
Args:
|
||||
host: Adresse IP ou hostname
|
||||
root_password: Mot de passe root
|
||||
automation_user: Nom de l'utilisateur à créer (défaut: settings.ssh_user)
|
||||
|
||||
Returns:
|
||||
CommandResult avec le résultat du bootstrap
|
||||
"""
|
||||
if not automation_user:
|
||||
automation_user = settings.ssh_user
|
||||
|
||||
# Trouver la clé publique
|
||||
public_key = find_ssh_public_key()
|
||||
if not public_key:
|
||||
return CommandResult(
|
||||
status="error",
|
||||
return_code=-1,
|
||||
stdout="",
|
||||
stderr="Aucune clé SSH publique trouvée. Générez une paire de clés SSH d'abord."
|
||||
)
|
||||
|
||||
# Script de bootstrap multi-OS (Debian/Ubuntu, Alpine, FreeBSD)
|
||||
bootstrap_script = f'''
|
||||
set -e
|
||||
|
||||
# Détection de l'OS
|
||||
if [ -f /etc/os-release ]; then
|
||||
. /etc/os-release
|
||||
OS_FAMILY="linux"
|
||||
if echo "$ID" | grep -qE "alpine"; then
|
||||
OS_TYPE="alpine"
|
||||
else
|
||||
OS_TYPE="debian"
|
||||
fi
|
||||
elif [ "$(uname)" = "FreeBSD" ]; then
|
||||
OS_FAMILY="bsd"
|
||||
OS_TYPE="freebsd"
|
||||
else
|
||||
OS_TYPE="unknown"
|
||||
fi
|
||||
|
||||
echo "=== OS détecté: $OS_TYPE ==="
|
||||
|
||||
# Créer l'utilisateur
|
||||
if ! id "{automation_user}" >/dev/null 2>&1; then
|
||||
echo "=== Création de l'utilisateur {automation_user} ==="
|
||||
if [ "$OS_TYPE" = "alpine" ]; then
|
||||
adduser -D -s /bin/sh "{automation_user}"
|
||||
elif [ "$OS_TYPE" = "freebsd" ]; then
|
||||
pw useradd "{automation_user}" -m -s /bin/sh
|
||||
else
|
||||
useradd -m -s /bin/bash "{automation_user}" || useradd -m -s /bin/sh "{automation_user}"
|
||||
fi
|
||||
fi
|
||||
|
||||
# Configurer SSH
|
||||
echo "=== Configuration SSH ==="
|
||||
USER_HOME=$(eval echo ~{automation_user})
|
||||
mkdir -p "$USER_HOME/.ssh"
|
||||
chmod 700 "$USER_HOME/.ssh"
|
||||
|
||||
# Ajouter la clé publique
|
||||
echo "{public_key}" >> "$USER_HOME/.ssh/authorized_keys"
|
||||
sort -u "$USER_HOME/.ssh/authorized_keys" -o "$USER_HOME/.ssh/authorized_keys"
|
||||
chmod 600 "$USER_HOME/.ssh/authorized_keys"
|
||||
|
||||
# Corriger les permissions
|
||||
if [ "$OS_TYPE" = "freebsd" ]; then
|
||||
chown -R "{automation_user}":"{automation_user}" "$USER_HOME/.ssh"
|
||||
else
|
||||
chown -R "{automation_user}":"{automation_user}" "$USER_HOME/.ssh" 2>/dev/null || chown -R "{automation_user}" "$USER_HOME/.ssh"
|
||||
fi
|
||||
|
||||
# Configurer sudo
|
||||
echo "=== Configuration sudo ==="
|
||||
if [ "$OS_TYPE" = "alpine" ]; then
|
||||
apk add --no-cache sudo 2>/dev/null || true
|
||||
elif [ "$OS_TYPE" = "freebsd" ]; then
|
||||
pkg install -y sudo 2>/dev/null || true
|
||||
fi
|
||||
|
||||
# Ajouter au groupe sudo/wheel
|
||||
if [ "$OS_TYPE" = "freebsd" ]; then
|
||||
pw groupmod wheel -m "{automation_user}" 2>/dev/null || true
|
||||
elif [ "$OS_TYPE" = "alpine" ]; then
|
||||
addgroup "{automation_user}" wheel 2>/dev/null || true
|
||||
else
|
||||
usermod -aG sudo "{automation_user}" 2>/dev/null || true
|
||||
fi
|
||||
|
||||
# Configurer sudoers
|
||||
SUDOERS_FILE="/etc/sudoers.d/{automation_user}"
|
||||
echo "{automation_user} ALL=(ALL) NOPASSWD: ALL" > "$SUDOERS_FILE"
|
||||
chmod 440 "$SUDOERS_FILE"
|
||||
|
||||
# Installer Python3
|
||||
echo "=== Installation Python3 ==="
|
||||
if ! command -v python3 >/dev/null 2>&1; then
|
||||
if [ "$OS_TYPE" = "alpine" ]; then
|
||||
apk add --no-cache python3
|
||||
elif [ "$OS_TYPE" = "freebsd" ]; then
|
||||
pkg install -y python3 || pkg install -y python39
|
||||
else
|
||||
apt-get update && apt-get install -y python3 || yum install -y python3
|
||||
fi
|
||||
fi
|
||||
|
||||
echo "=== Bootstrap terminé avec succès ==="
|
||||
'''
|
||||
|
||||
# Exécuter le script de bootstrap
|
||||
print(f"[Bootstrap] Connexion à {host}...")
|
||||
result = run_ssh_command(
|
||||
host=host,
|
||||
command=bootstrap_script,
|
||||
user="root",
|
||||
password=root_password,
|
||||
timeout=120
|
||||
)
|
||||
|
||||
if result.return_code != 0:
|
||||
return result
|
||||
|
||||
# Vérifier la connexion SSH par clé
|
||||
print(f"[Bootstrap] Vérification de la connexion par clé SSH...")
|
||||
verify_result = run_ssh_command(
|
||||
host=host,
|
||||
command="echo 'SSH key authentication OK'",
|
||||
user=automation_user,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
if verify_result.return_code != 0:
|
||||
return CommandResult(
|
||||
status="partial",
|
||||
return_code=1,
|
||||
stdout=result.stdout,
|
||||
stderr=f"Bootstrap terminé mais la vérification SSH a échoué: {verify_result.stderr}"
|
||||
)
|
||||
|
||||
return CommandResult(
|
||||
status="success",
|
||||
return_code=0,
|
||||
stdout=result.stdout + "\n" + verify_result.stdout,
|
||||
stderr=result.stderr
|
||||
)
|
||||
Reference in New Issue
Block a user