gilles/LANMap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move

Browse Source
This commit is contained in:
gilles soulier
2025-12-07 01:16:52 +01:00
parent 13b3c58ec8
commit 78e6909405
22 changed files with 4664 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
backend/app/core/config.py
View File

@@ -33,6 +33,7 @@ class ScanConfig(BaseModel):
class PortsConfig(BaseModel):
"""Configuration des ports à scanner"""
ranges: List[str] = ["22", "80", "443", "3389", "8080"]
protocols: Optional[Dict[int, str]] = None # Mapping port -> protocole
class HistoryConfig(BaseModel):
@@ -45,6 +46,11 @@ class UIConfig(BaseModel):
offline_transparency: float = 0.5
show_mac: bool = True
show_vendor: bool = True
cell_size: int = 30
font_size: int = 10
cell_gap: float = 2
details_font_size: int = 13
details_spacing: int = 2
class ColorsConfig(BaseModel):
@@ -61,15 +67,23 @@ class DatabaseConfig(BaseModel):
path: str = "./data/db.sqlite"
class HostConfig(BaseModel):
"""Configuration d'un hôte avec sa localisation"""
name: str
location: str
class IPWatchConfig(BaseModel):
"""Configuration complète IPWatch"""
model_config = {"arbitrary_types_allowed": True}
app: AppConfig = Field(default_factory=AppConfig)
network: NetworkConfig
ip_classes: Dict[str, Any] = Field(default_factory=dict)
scan: ScanConfig = Field(default_factory=ScanConfig)
ports: PortsConfig = Field(default_factory=PortsConfig)
locations: List[str] = Field(default_factory=list)
hosts: List[str] = Field(default_factory=list)
hosts: List[HostConfig] = Field(default_factory=list)
history: HistoryConfig = Field(default_factory=HistoryConfig)
ui: UIConfig = Field(default_factory=UIConfig)
colors: ColorsConfig = Field(default_factory=ColorsConfig)
@@ -97,8 +111,15 @@ class ConfigManager:
yaml_data = yaml.safe_load(f)
self._config = IPWatchConfig(**yaml_data)
self._config_path = config_path
return self._config
def reload_config(self) -> IPWatchConfig:
"""Recharge la configuration depuis le fichier"""
if not hasattr(self, '_config_path'):
self._config_path = "./config.yaml"
return self.load_config(self._config_path)
@property
def config(self) -> IPWatchConfig:
"""Retourne la configuration actuelle"""

2
backend/app/main.py
View File

@@ -12,6 +12,7 @@ from pathlib import Path
from backend.app.core.config import config_manager
from backend.app.core.database import init_database, get_db
from backend.app.routers import ips_router, scan_router, websocket_router
from backend.app.routers import config as config_router
from backend.app.services.scheduler import scan_scheduler
from backend.app.routers.scan import perform_scan
@@ -119,6 +120,7 @@ app.add_middleware(
app.include_router(ips_router)
app.include_router(scan_router)
app.include_router(websocket_router)
app.include_router(config_router.router)
@app.get("/health")

1
backend/app/models/ip.py
View File

@@ -35,6 +35,7 @@ class IP(Base):
mac = Column(String, nullable=True) # Adresse MAC
vendor = Column(String, nullable=True) # Fabricant (lookup MAC)
hostname = Column(String, nullable=True) # Nom d'hôte réseau
link = Column(String, nullable=True) # Lien personnalisé (URL)
# Ports ouverts (stocké en JSON)
open_ports = Column(JSON, default=list) # Liste des ports ouverts

40
backend/app/routers/config.py Normal file
View File

@@ -0,0 +1,40 @@
"""
Routes pour la configuration
"""
from fastapi import APIRouter, HTTPException
from backend.app.core.config import config_manager
router = APIRouter(prefix="/api/config", tags=["config"])
@router.get("/ui")
async def get_ui_config():
"""Récupérer la configuration UI"""
config = config_manager.config
return {
"cell_size": config.ui.cell_size,
"font_size": config.ui.font_size,
"cell_gap": config.ui.cell_gap,
"offline_transparency": config.ui.offline_transparency,
"show_mac": config.ui.show_mac,
"show_vendor": config.ui.show_vendor
}
@router.post("/reload")
async def reload_config():
"""Recharger la configuration depuis le fichier config.yaml"""
try:
config = config_manager.reload_config()
return {
"success": True,
"message": "Configuration rechargée avec succès",
"ui": {
"cell_size": config.ui.cell_size,
"font_size": config.ui.font_size,
"cell_gap": config.ui.cell_gap,
"offline_transparency": config.ui.offline_transparency,
"show_mac": config.ui.show_mac,
"show_vendor": config.ui.show_vendor
}
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erreur rechargement config: {str(e)}")

193
backend/app/routers/ips.py
View File

@@ -1,14 +1,17 @@
"""
Endpoints API pour la gestion des IPs
"""
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
from sqlalchemy.orm import Session
from sqlalchemy import desc
from typing import List, Optional
from datetime import datetime, timedelta
import xml.etree.ElementTree as ET
import yaml
from backend.app.core.database import get_db
from backend.app.models.ip import IP, IPHistory
from backend.app.core.config import config_manager
from pydantic import BaseModel
router = APIRouter(prefix="/api/ips", tags=["IPs"])
@@ -21,6 +24,7 @@ class IPUpdate(BaseModel):
known: Optional[bool] = None
location: Optional[str] = None
host: Optional[str] = None
link: Optional[str] = None
class IPResponse(BaseModel):
@@ -36,6 +40,7 @@ class IPResponse(BaseModel):
mac: Optional[str]
vendor: Optional[str]
hostname: Optional[str]
link: Optional[str]
open_ports: List[int]
class Config:
@@ -214,3 +219,189 @@ async def get_stats(db: Session = Depends(get_db)):
"known": known,
"unknown": unknown
}
@router.get("/config/options")
async def get_config_options():
"""
Récupère les options de configuration (locations, hosts, port_protocols, version)
Returns:
Dictionnaire avec locations, hosts, port_protocols et version
"""
config = config_manager.config
# Récupérer les protocoles de ports depuis la config
port_protocols = {}
if hasattr(config.ports, 'protocols') and config.ports.protocols:
port_protocols = config.ports.protocols
return {
"locations": config.locations,
"hosts": [{"name": h.name, "location": h.location} for h in config.hosts],
"port_protocols": port_protocols,
"version": config.app.version
}
@router.get("/config/content")
async def get_config_content():
"""
Récupère le contenu brut du fichier config.yaml
Returns:
Contenu du fichier YAML
"""
try:
with open("./config.yaml", "r", encoding="utf-8") as f:
content = f.read()
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erreur lecture config: {str(e)}")
@router.post("/config/reload")
async def reload_config():
"""
Recharge la configuration depuis config.yaml
Returns:
Message de confirmation
"""
try:
config_manager.reload_config()
return {"message": "Configuration rechargée avec succès"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erreur rechargement config: {str(e)}")
@router.post("/import/ipscan")
async def import_ipscan(file: UploadFile = File(...), db: Session = Depends(get_db)):
"""
Importe les données depuis un fichier XML Angry IP Scanner
Args:
file: Fichier XML uploadé
db: Session de base de données
Returns:
Statistiques d'import
"""
if not file.filename.endswith('.xml'):
raise HTTPException(status_code=400, detail="Le fichier doit être un XML")
try:
# Lire le contenu du fichier
content = await file.read()
# Essayer de parser le XML avec récupération d'erreurs
try:
root = ET.fromstring(content)
except ET.ParseError as e:
# Si le parsing échoue, essayer de nettoyer le contenu
import re
content_str = content.decode('utf-8', errors='ignore')
# Supprimer les caractères de contrôle invalides (sauf tab, CR, LF)
content_str = ''.join(char for char in content_str
if ord(char) >= 32 or char in '\t\r\n')
try:
root = ET.fromstring(content_str.encode('utf-8'))
except ET.ParseError:
raise HTTPException(status_code=400, detail=f"Fichier XML invalide même après nettoyage: {str(e)}")
imported = 0
updated = 0
errors = []
# Parser chaque host
for host in root.findall('.//host'):
try:
# Extraire l'adresse IP
ip_address = host.get('address')
if not ip_address:
continue
# Extraire les informations
hostname = None
mac = None
vendor = None
ports = []
for result in host.findall('result'):
name = result.get('name')
value = result.text.strip() if result.text else ""
# Nettoyer les valeurs [n/a]
if value == "[n/a]":
value = None
if name == "Nom d'hôte" and value:
hostname = value
elif name == "Adresse MAC" and value:
mac = value
elif name == "Constructeur MAC" and value:
vendor = value
elif name == "Ports" and value:
# Parser les ports (format: "22,80,443")
try:
ports = [int(p.strip()) for p in value.split(',') if p.strip().isdigit()]
except Exception as e:
ports = []
# Vérifier si l'IP existe déjà
existing_ip = db.query(IP).filter(IP.ip == ip_address).first()
if existing_ip:
# Mettre à jour avec de nouvelles informations
if hostname:
if not existing_ip.hostname:
existing_ip.hostname = hostname
if not existing_ip.name:
existing_ip.name = hostname
if mac and not existing_ip.mac:
existing_ip.mac = mac
# Toujours mettre à jour vendor et ports depuis IPScan (plus complet et à jour)
if vendor:
existing_ip.vendor = vendor
if ports:
existing_ip.open_ports = ports
existing_ip.last_status = "online"
existing_ip.last_seen = datetime.utcnow()
updated += 1
else:
# Créer une nouvelle entrée
new_ip = IP(
ip=ip_address,
name=hostname,
hostname=hostname,
mac=mac,
vendor=vendor,
open_ports=ports or [],
last_status="online",
known=False,
first_seen=datetime.utcnow(),
last_seen=datetime.utcnow()
)
db.add(new_ip)
imported += 1
except Exception as e:
errors.append(f"Erreur pour {ip_address}: {str(e)}")
continue
# Commit des changements
db.commit()
return {
"message": "Import terminé",
"imported": imported,
"updated": updated,
"errors": errors[:10] # Limiter à 10 erreurs
}
except ET.ParseError as e:
raise HTTPException(status_code=400, detail=f"Fichier XML invalide: {str(e)}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erreur import: {str(e)}")

66
backend/app/routers/scan.py
View File

@@ -174,6 +174,72 @@ async def start_scan(background_tasks: BackgroundTasks, db: Session = Depends(ge
}
@router.post("/ports/{ip_address}")
async def scan_ip_ports(ip_address: str, db: Session = Depends(get_db)):
"""
Scanne les ports d'une IP spécifique
Args:
ip_address: Adresse IP à scanner
db: Session de base de données
Returns:
Liste des ports ouverts
"""
try:
# Récupérer la config
config = config_manager.config
# Convertir les ports en liste d'entiers
port_list = []
for port_range in config.ports.ranges:
if '-' in port_range:
start, end = map(int, port_range.split('-'))
port_list.extend(range(start, end + 1))
else:
port_list.append(int(port_range))
# Initialiser le scanner
scanner = NetworkScanner(
cidr=config.network.cidr,
timeout=config.scan.timeout
)
# Scanner les ports de cette IP
print(f"[{datetime.now()}] Scan ports pour {ip_address}...")
open_ports = await scanner.scan_ports(ip_address, port_list)
print(f"[{datetime.now()}] Ports ouverts pour {ip_address}: {open_ports}")
# Mettre à jour la base de données
ip_record = db.query(IP).filter(IP.ip == ip_address).first()
if ip_record:
ip_record.open_ports = open_ports
ip_record.last_seen = datetime.utcnow()
db.commit()
# Notifier via WebSocket
await ws_manager.broadcast_ip_update({
"ip": ip_address,
"open_ports": open_ports
})
return {
"message": "Scan de ports terminé",
"ip": ip_address,
"open_ports": open_ports,
"timestamp": datetime.utcnow()
}
except Exception as e:
print(f"Erreur scan ports {ip_address}: {e}")
return {
"message": f"Erreur: {str(e)}",
"ip": ip_address,
"open_ports": [],
"timestamp": datetime.utcnow()
}
@router.post("/cleanup-history")
async def cleanup_history(hours: int = 24, db: Session = Depends(get_db)):
"""