Claude Add complete API files with filesystem endpoints

This commit is contained in:
nicoboy
2026-01-01 17:47:52 +01:00
parent 9550e05045
commit 7e72663685
11 changed files with 1268 additions and 0 deletions

212
README.md Normal file
View File

@ -0,0 +1,212 @@
# Jetson Agent API
API middleware sécurisée pour accès lecture aux fichiers du Jetson Nano.
## 📋 Description
Cette API permet un accès sécurisé et contrôlé aux fichiers du Jetson Nano, principalement pour permettre à des LLMs (Claude, Gemini, etc.) de lire et analyser les fichiers de l'application web mathématiques.
## ✨ Fonctionnalités
-**Lecture de fichiers** sécurisée avec validation de chemins
-**Listing de répertoires** avec support récursif
-**Arborescence** (tree) de dossiers
-**Recherche** (grep) dans les fichiers
-**Validation** d'extensions et tailles de fichiers
-**API REST** avec documentation Swagger intégrée
## 🚀 Installation
### Prérequis
- Python 3.11+
- pip
- Git
### Installation rapide
```bash
# Se positionner dans le projet
cd ~/projects/jetson-agent
# Créer environnement virtuel
python3 -m venv .venv
# Activer l'environnement
source .venv/bin/activate
# Installer les dépendances
pip install -r requirements.txt
```
## ⚙️ Configuration
### 1. Créer le fichier .env
```bash
# Copier l'exemple
cp config/example.env .env
# Générer une clé secrète JWT
openssl rand -hex 32
# Éditer .env et remplacer JWT_SECRET_KEY par la clé générée
nano .env
```
### 2. Adapter les chemins autorisés
Éditer `config/allowed_paths.yaml` pour ajouter/retirer des chemins :
```yaml
allowed_read_paths:
- /var/www/mathematiques
- /var/www/html
# Ajouter d'autres chemins si nécessaire
```
## 🏃 Lancement
### Mode développement
```bash
# Activer l'environnement
source .venv/bin/activate
# Lancer l'API
cd src
uvicorn jetson_agent.main:app --reload --host 0.0.0.0 --port 8000
```
L'API est accessible sur : `http://localhost:8000`
Documentation Swagger : `http://localhost:8000/docs`
### Mode production (avec systemd)
Créer un service systemd (à venir).
## 📚 Utilisation de l'API
### Endpoints disponibles
#### 1. Lister les fichiers d'un dossier
```bash
GET /api/files/list?path=/var/www/mathematiques
# Avec récursion
GET /api/files/list?path=/var/www/mathematiques&recursive=true&max_depth=3
```
#### 2. Lire un fichier
```bash
GET /api/files/read?path=/var/www/mathematiques/index.php
```
#### 3. Arborescence (tree)
```bash
GET /api/files/tree?path=/var/www/mathematiques&max_depth=5
```
#### 4. Rechercher dans les fichiers (grep)
```bash
GET /api/files/grep?path=/var/www/mathematiques&pattern=function&file_pattern=*.php
```
### Exemples avec curl
```bash
# Health check
curl http://localhost:8000/api/health
# Lister fichiers
curl "http://localhost:8000/api/files/list?path=/var/www/mathematiques"
# Lire un fichier
curl "http://localhost:8000/api/files/read?path=/var/www/mathematiques/index.php"
# Arborescence
curl "http://localhost:8000/api/files/tree?path=/var/www/mathematiques&max_depth=3"
# Recherche
curl "http://localhost:8000/api/files/grep?path=/var/www/mathematiques&pattern=mysql&file_pattern=*.php"
```
## 🔐 Sécurité
### Chemins autorisés
L'API n'autorise l'accès qu'aux chemins définis dans `config/allowed_paths.yaml`.
### Validation
- ✅ Pas de traversal de chemin (`..`)
- ✅ Blocklist de chemins sensibles (`/etc`, `/root`, etc.)
- ✅ Validation d'extensions de fichiers
- ✅ Limite de taille de fichiers (10 MB par défaut)
- ✅ Limite du nombre de résultats
### Recommandations
- [ ] Utiliser HTTPS en production (via nginx reverse proxy)
- [ ] Configurer un pare-feu (UFW)
- [ ] Limiter l'accès IP si possible
- [ ] Activer l'authentification JWT (à venir)
## 📁 Structure du projet
```
jetson-agent/
├── src/
│ └── jetson_agent/
│ ├── __init__.py
│ ├── main.py # Point d'entrée FastAPI
│ ├── config.py # Configuration
│ ├── path_validator.py # Validation de chemins
│ └── api/
│ ├── __init__.py
│ └── filesystem.py # Endpoints fichiers
├── config/
│ ├── example.env
│ └── allowed_paths.yaml
├── tests/ # Tests (à venir)
├── requirements.txt
└── README.md
```
## 🧪 Tests
```bash
# Installer les dépendances de dev
pip install pytest pytest-asyncio httpx
# Lancer les tests (à venir)
pytest
```
## 📝 TODO / Roadmap
- [ ] Authentification JWT
- [ ] Endpoints base de données (lecture MariaDB)
- [ ] Rate limiting
- [ ] Logs structurés
- [ ] Tests unitaires
- [ ] Déploiement systemd
- [ ] Documentation API complète
- [ ] Support GPIO (lecture/écriture pins)
## 🤝 Contribution
Projet personnel pour contrôle du Jetson Nano via LLM.
## 📄 Licence
Privé - Usage personnel
## 👤 Auteur
Nicolas - 2025

50
config/allowed_paths.yaml Normal file
View File

@ -0,0 +1,50 @@
# Configuration des chemins autorisés pour l'accès fichiers
# Format YAML
# Chemins autorisés en LECTURE
allowed_read_paths:
- /var/www/mathematiques
- /var/www/html
- /enseignement
# Chemins autorisés en ÉCRITURE (pour plus tard)
allowed_write_paths:
- /var/www/mathematiques/_generated
- /var/www/mathematiques/uploads
# Chemins BLOQUÉS (sécurité)
blocked_paths:
- /etc
- /root
- /home
- /boot
- /sys
- /proc
- /.ssh
- /var/log
# Extensions de fichiers autorisées
allowed_extensions:
- .php
- .html
- .css
- .js
- .json
- .xml
- .txt
- .md
- .sql
- .py
- .sh
- .conf
- .yaml
- .yml
# Taille maximale de fichier (en octets)
max_file_size: 10485760 # 10 MB
# Nombre max de fichiers retournés par listing
max_files_listing: 1000
# Profondeur max pour récursion
max_recursion_depth: 10

34
config/example.env Normal file
View File

@ -0,0 +1,34 @@
# Exemple de configuration - Copier vers .env et modifier les valeurs
# Application
APP_NAME="Jetson Agent API"
APP_VERSION="1.0.0"
APP_ENV="development" # development, production
LOG_LEVEL="info" # debug, info, warning, error
# API Configuration
API_HOST="0.0.0.0"
API_PORT=8000
API_RELOAD=true # false en production
# Sécurité JWT
JWT_SECRET_KEY="CHANGE_ME_GENERATE_WITH_openssl_rand_hex_32"
JWT_ALGORITHM="HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=1440 # 24 heures
# Chemins de configuration
CONFIG_DIR="/home/nicoboy/projects/jetson-agent/config"
ALLOWED_PATHS_CONFIG="allowed_paths.yaml"
# Réseau et sécurité
ALLOWED_IPS="192.168.1.0/24,82.67.167.0/24" # Adapte selon ton réseau
CORS_ORIGINS="http://localhost:3000,https://82.67.167.147"
# Rate limiting
RATE_LIMIT_ENABLED=true
RATE_LIMIT_PER_MINUTE=100
# Logs
LOG_FILE="/home/nicoboy/projects/jetson-agent/logs/api.log"
LOG_ROTATION_SIZE=10485760 # 10 MB
LOG_BACKUP_COUNT=5

20
requirements.txt Normal file
View File

@ -0,0 +1,20 @@
# FastAPI et serveur
fastapi==0.109.0
uvicorn[standard]==0.27.0
python-multipart==0.0.6
# Sécurité et auth
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
pydantic==2.5.3
pydantic-settings==2.1.0
# Utilitaires
aiofiles==23.2.1
python-dateutil==2.8.2
pyyaml==6.0.1
# Développement (optionnel)
pytest==7.4.3
pytest-asyncio==0.21.1
httpx==0.26.0

109
setup.sh Executable file
View File

@ -0,0 +1,109 @@
#!/bin/bash
#
# Script d'installation et configuration de Jetson Agent API
# Usage: ./setup.sh
#
set -e # Arrêter en cas d'erreur
echo "🚀 Installation de Jetson Agent API"
echo "===================================="
echo ""
# Couleurs pour l'affichage
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# Vérifier qu'on est dans le bon dossier
if [ ! -f "requirements.txt" ]; then
echo "❌ Erreur: requirements.txt non trouvé"
echo " Assurez-vous d'être dans le dossier ~/projects/jetson-agent"
exit 1
fi
echo "📦 Étape 1: Création de l'environnement virtuel Python"
echo "-------------------------------------------------------"
# Créer venv s'il n'existe pas
if [ ! -d ".venv" ]; then
python3 -m venv .venv
echo -e "${GREEN}✅ Environnement virtuel créé${NC}"
else
echo -e "${YELLOW}⚠️ Environnement virtuel existe déjà${NC}"
fi
# Activer venv
source .venv/bin/activate
echo -e "${GREEN}✅ Environnement virtuel activé${NC}"
echo ""
echo "📚 Étape 2: Installation des dépendances Python"
echo "-----------------------------------------------"
pip install --upgrade pip
pip install -r requirements.txt
echo -e "${GREEN}✅ Dépendances installées${NC}"
echo ""
echo "⚙️ Étape 3: Configuration"
echo "-------------------------"
# Créer dossier config s'il n'existe pas
mkdir -p config
mkdir -p logs
# Copier example.env vers .env si pas existant
if [ ! -f ".env" ]; then
if [ -f "config/example.env" ]; then
cp config/example.env .env
echo -e "${GREEN}✅ Fichier .env créé${NC}"
# Générer une clé JWT secrète
JWT_SECRET=$(openssl rand -hex 32)
sed -i "s/CHANGE_ME_GENERATE_WITH_openssl_rand_hex_32/$JWT_SECRET/" .env
echo -e "${GREEN}✅ Clé JWT générée${NC}"
else
echo -e "${YELLOW}⚠️ example.env non trouvé, .env non créé${NC}"
fi
else
echo -e "${YELLOW}⚠️ Fichier .env existe déjà (non modifié)${NC}"
fi
# Vérifier allowed_paths.yaml
if [ ! -f "config/allowed_paths.yaml" ]; then
echo -e "${YELLOW}⚠️ config/allowed_paths.yaml manquant${NC}"
else
echo -e "${GREEN}✅ config/allowed_paths.yaml présent${NC}"
fi
echo ""
echo "🧪 Étape 4: Test de l'installation"
echo "----------------------------------"
# Tester l'import Python
python3 -c "import fastapi; print('✅ FastAPI OK')" 2>/dev/null || echo "❌ FastAPI manquant"
python3 -c "import uvicorn; print('✅ Uvicorn OK')" 2>/dev/null || echo "❌ Uvicorn manquant"
python3 -c "import pydantic; print('✅ Pydantic OK')" 2>/dev/null || echo "❌ Pydantic manquant"
python3 -c "import yaml; print('✅ PyYAML OK')" 2>/dev/null || echo "❌ PyYAML manquant"
echo ""
echo "✨ Installation terminée !"
echo "=========================="
echo ""
echo "📝 Prochaines étapes:"
echo ""
echo "1. Vérifier/adapter la configuration:"
echo " nano .env"
echo " nano config/allowed_paths.yaml"
echo ""
echo "2. Lancer l'API en mode développement:"
echo " source .venv/bin/activate"
echo " cd src"
echo " uvicorn jetson_agent.main:app --reload --host 0.0.0.0 --port 8000"
echo ""
echo "3. Accéder à la documentation:"
echo " http://localhost:8000/docs"
echo ""
echo -e "${GREEN}🎉 Prêt à démarrer !${NC}"

View File

@ -0,0 +1,7 @@
"""
Jetson Agent API Package
Secure middleware API for Jetson Nano control
"""
__version__ = "1.0.0"
__author__ = "Nicolas"

View File

@ -0,0 +1,4 @@
"""
API package
Contains all API endpoint routers
"""

View File

@ -0,0 +1,421 @@
"""
Filesystem API endpoints
Provides secure read access to files and directories
"""
from pathlib import Path
from typing import List, Optional
from datetime import datetime
import os
import stat
import re
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel, Field
from ..path_validator import path_validator
from ..config import paths_config
router = APIRouter(prefix="/files", tags=["filesystem"])
# ============================================================================
# Pydantic Models
# ============================================================================
class FileInfo(BaseModel):
"""Information about a file or directory"""
path: str
name: str
type: str = Field(description="file, directory, or symlink")
size: Optional[int] = None
modified: Optional[datetime] = None
permissions: Optional[str] = None
owner: Optional[str] = None
is_readable: bool = True
is_text: bool = False
class ListFilesResponse(BaseModel):
"""Response for file listing"""
path: str
files: List[FileInfo]
total_count: int
total_size: int
class ReadFileResponse(BaseModel):
"""Response for file content"""
path: str
content: str
size: int
encoding: str = "utf-8"
is_text: bool = True
class GrepMatch(BaseModel):
"""A single grep match result"""
file: str
line_number: int
line_content: str
class GrepResponse(BaseModel):
"""Response for grep search"""
pattern: str
path: str
matches: List[GrepMatch]
total_matches: int
files_searched: int
# ============================================================================
# Helper Functions
# ============================================================================
def get_file_info(file_path: Path) -> FileInfo:
"""Get detailed information about a file"""
try:
file_stat = file_path.stat()
# Déterminer le type
if file_path.is_symlink():
file_type = "symlink"
elif file_path.is_dir():
file_type = "directory"
else:
file_type = "file"
# Permissions en format rwxrwxrwx
mode = file_stat.st_mode
permissions = stat.filemode(mode)
# Propriétaire (si possible)
try:
owner = f"{file_stat.st_uid}"
except Exception:
owner = "unknown"
return FileInfo(
path=str(file_path),
name=file_path.name,
type=file_type,
size=file_stat.st_size if file_type == "file" else None,
modified=datetime.fromtimestamp(file_stat.st_mtime),
permissions=permissions,
owner=owner,
is_readable=os.access(file_path, os.R_OK),
is_text=path_validator.is_text_file(file_path) if file_type == "file" else False
)
except Exception as e:
# Si erreur, retourner info minimale
return FileInfo(
path=str(file_path),
name=file_path.name,
type="unknown",
is_readable=False
)
# ============================================================================
# Endpoints
# ============================================================================
@router.get("/list", response_model=ListFilesResponse)
async def list_files(
path: str = Query(..., description="Directory path to list"),
recursive: bool = Query(False, description="List files recursively"),
max_depth: int = Query(3, ge=1, le=10, description="Maximum recursion depth")
):
"""
List files in a directory
- **path**: Directory path (must be in allowed paths)
- **recursive**: If true, list files recursively
- **max_depth**: Maximum depth for recursive listing
"""
# Valider le chemin
dir_path = path_validator.validate_read_path(path)
if not dir_path.is_dir():
raise HTTPException(
status_code=400,
detail=f"Path is not a directory: {path}"
)
files_list = []
total_size = 0
def list_recursive(current_path: Path, current_depth: int = 0):
"""Recursively list files"""
nonlocal total_size
if current_depth > max_depth:
return
try:
for item in current_path.iterdir():
# Vérifier limite de fichiers
if len(files_list) >= paths_config.max_files_listing:
return
try:
file_info = get_file_info(item)
files_list.append(file_info)
if file_info.size:
total_size += file_info.size
# Récursion si demandé et si c'est un dossier
if recursive and item.is_dir():
list_recursive(item, current_depth + 1)
except PermissionError:
continue
except Exception:
continue
except PermissionError:
raise HTTPException(
status_code=403,
detail=f"Permission denied: {current_path}"
)
# Lister les fichiers
list_recursive(dir_path)
return ListFilesResponse(
path=str(dir_path),
files=files_list,
total_count=len(files_list),
total_size=total_size
)
@router.get("/read", response_model=ReadFileResponse)
async def read_file(
path: str = Query(..., description="File path to read"),
encoding: str = Query("utf-8", description="File encoding"),
max_size: Optional[int] = Query(None, description="Maximum file size in bytes")
):
"""
Read the content of a text file
- **path**: File path (must be in allowed paths)
- **encoding**: Text encoding (utf-8, latin-1, etc.)
- **max_size**: Override maximum file size (capped at config limit)
"""
# Valider le chemin
file_path = path_validator.validate_read_path(path)
if not file_path.is_file():
raise HTTPException(
status_code=400,
detail=f"Path is not a file: {path}"
)
# Vérifier l'extension
if not path_validator.validate_file_extension(file_path):
raise HTTPException(
status_code=403,
detail=f"File extension not allowed: {file_path.suffix}"
)
# Vérifier la taille
is_valid_size, file_size = path_validator.validate_file_size(file_path)
# Appliquer max_size si fourni (mais limité par config)
effective_max_size = min(
max_size or paths_config.max_file_size,
paths_config.max_file_size
)
if file_size > effective_max_size:
raise HTTPException(
status_code=413,
detail=f"File too large: {file_size} bytes (max: {effective_max_size} bytes)"
)
# Lire le fichier
try:
with open(file_path, 'r', encoding=encoding) as f:
content = f.read()
return ReadFileResponse(
path=str(file_path),
content=content,
size=file_size,
encoding=encoding,
is_text=path_validator.is_text_file(file_path)
)
except UnicodeDecodeError:
raise HTTPException(
status_code=400,
detail=f"Cannot decode file with encoding '{encoding}'. Try 'latin-1' or 'binary'."
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Error reading file: {str(e)}"
)
@router.get("/tree")
async def get_tree(
path: str = Query(..., description="Directory path"),
max_depth: int = Query(5, ge=1, le=10, description="Maximum depth")
):
"""
Get directory tree structure as text
- **path**: Directory path
- **max_depth**: Maximum depth for tree
"""
# Valider le chemin
dir_path = path_validator.validate_read_path(path)
if not dir_path.is_dir():
raise HTTPException(
status_code=400,
detail=f"Path is not a directory: {path}"
)
tree_lines = []
total_files = 0
total_dirs = 0
total_size = 0
def build_tree(current_path: Path, prefix: str = "", current_depth: int = 0):
"""Build tree recursively"""
nonlocal total_files, total_dirs, total_size
if current_depth > max_depth:
return
try:
items = sorted(current_path.iterdir(), key=lambda x: (not x.is_dir(), x.name))
for i, item in enumerate(items):
is_last = i == len(items) - 1
current_prefix = "└── " if is_last else "├── "
next_prefix = " " if is_last else ""
try:
if item.is_dir():
tree_lines.append(f"{prefix}{current_prefix}{item.name}/")
total_dirs += 1
build_tree(item, prefix + next_prefix, current_depth + 1)
else:
size = item.stat().st_size
total_files += 1
total_size += size
size_str = f"{size:,} bytes" if size < 1024 else f"{size/1024:.1f} KB"
tree_lines.append(f"{prefix}{current_prefix}{item.name} ({size_str})")
except PermissionError:
tree_lines.append(f"{prefix}{current_prefix}{item.name} [Permission Denied]")
except Exception:
continue
except PermissionError:
tree_lines.append(f"{prefix}[Permission Denied]")
# Construire l'arbre
tree_lines.append(f"{dir_path}/")
build_tree(dir_path)
tree_text = "\n".join(tree_lines)
return {
"path": str(dir_path),
"tree": tree_text,
"stats": {
"total_files": total_files,
"total_dirs": total_dirs,
"total_size": total_size
}
}
@router.get("/grep", response_model=GrepResponse)
async def grep_files(
path: str = Query(..., description="Directory to search in"),
pattern: str = Query(..., description="Search pattern (regex)"),
file_pattern: str = Query("*", description="File pattern (e.g., *.php)"),
case_sensitive: bool = Query(False, description="Case sensitive search"),
max_results: int = Query(100, ge=1, le=1000, description="Maximum results")
):
"""
Search for a pattern in files (like grep)
- **path**: Directory to search in
- **pattern**: Regular expression pattern to search
- **file_pattern**: File glob pattern (e.g., *.php, *.txt)
- **case_sensitive**: Case sensitive search
- **max_results**: Maximum number of results
"""
# Valider le chemin
dir_path = path_validator.validate_read_path(path)
if not dir_path.is_dir():
raise HTTPException(
status_code=400,
detail=f"Path is not a directory: {path}"
)
# Compiler le pattern de recherche
try:
regex_flags = 0 if case_sensitive else re.IGNORECASE
compiled_pattern = re.compile(pattern, regex_flags)
except re.error as e:
raise HTTPException(
status_code=400,
detail=f"Invalid regex pattern: {str(e)}"
)
matches = []
files_searched = 0
# Parcourir les fichiers
for file_path in dir_path.rglob(file_pattern):
if not file_path.is_file():
continue
# Vérifier extension
if not path_validator.validate_file_extension(file_path):
continue
# Vérifier si fichier texte
if not path_validator.is_text_file(file_path):
continue
files_searched += 1
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
for line_num, line in enumerate(f, 1):
if compiled_pattern.search(line):
matches.append(GrepMatch(
file=str(file_path),
line_number=line_num,
line_content=line.rstrip()
))
# Limiter les résultats
if len(matches) >= max_results:
break
if len(matches) >= max_results:
break
except Exception:
continue
return GrepResponse(
pattern=pattern,
path=str(dir_path),
matches=matches,
total_matches=len(matches),
files_searched=files_searched
)

124
src/jetson_agent/config.py Normal file
View File

@ -0,0 +1,124 @@
"""
Configuration module for Jetson Agent API
Loads settings from environment variables and YAML files
"""
from pathlib import Path
from typing import List, Optional
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
import yaml
class Settings(BaseSettings):
"""Application settings loaded from environment variables"""
# Application
app_name: str = Field(default="Jetson Agent API", alias="APP_NAME")
app_version: str = Field(default="1.0.0", alias="APP_VERSION")
app_env: str = Field(default="development", alias="APP_ENV")
log_level: str = Field(default="info", alias="LOG_LEVEL")
# API Configuration
api_host: str = Field(default="0.0.0.0", alias="API_HOST")
api_port: int = Field(default=8000, alias="API_PORT")
api_reload: bool = Field(default=True, alias="API_RELOAD")
# Security - JWT
jwt_secret_key: str = Field(default="INSECURE_CHANGE_ME", alias="JWT_SECRET_KEY")
jwt_algorithm: str = Field(default="HS256", alias="JWT_ALGORITHM")
jwt_access_token_expire_minutes: int = Field(default=1440, alias="JWT_ACCESS_TOKEN_EXPIRE_MINUTES")
# Paths
config_dir: Path = Field(default=Path("/home/nicoboy/projects/jetson-agent/config"), alias="CONFIG_DIR")
allowed_paths_config: str = Field(default="allowed_paths.yaml", alias="ALLOWED_PATHS_CONFIG")
# Network
allowed_ips: str = Field(default="192.168.1.0/24", alias="ALLOWED_IPS")
cors_origins: str = Field(default="http://localhost:3000", alias="CORS_ORIGINS")
# Rate limiting
rate_limit_enabled: bool = Field(default=True, alias="RATE_LIMIT_ENABLED")
rate_limit_per_minute: int = Field(default=100, alias="RATE_LIMIT_PER_MINUTE")
# Logging
log_file: Optional[Path] = Field(default=None, alias="LOG_FILE")
log_rotation_size: int = Field(default=10485760, alias="LOG_ROTATION_SIZE")
log_backup_count: int = Field(default=5, alias="LOG_BACKUP_COUNT")
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore"
)
@property
def allowed_ips_list(self) -> List[str]:
"""Convert comma-separated IPs to list"""
return [ip.strip() for ip in self.allowed_ips.split(",")]
@property
def cors_origins_list(self) -> List[str]:
"""Convert comma-separated CORS origins to list"""
return [origin.strip() for origin in self.cors_origins.split(",")]
class PathsConfig:
"""Configuration for allowed/blocked paths loaded from YAML"""
def __init__(self, config_path: Path):
self.config_path = config_path
self._config = self._load_config()
def _load_config(self) -> dict:
"""Load configuration from YAML file"""
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
except FileNotFoundError:
# Retourner config par défaut si fichier absent
return {
'allowed_read_paths': ['/var/www/mathematiques'],
'allowed_write_paths': [],
'blocked_paths': ['/etc', '/root', '/home', '/boot'],
'allowed_extensions': ['.php', '.html', '.txt', '.md'],
'max_file_size': 10485760,
'max_files_listing': 1000,
'max_recursion_depth': 10
}
@property
def allowed_read_paths(self) -> List[str]:
return self._config.get('allowed_read_paths', [])
@property
def allowed_write_paths(self) -> List[str]:
return self._config.get('allowed_write_paths', [])
@property
def blocked_paths(self) -> List[str]:
return self._config.get('blocked_paths', [])
@property
def allowed_extensions(self) -> List[str]:
return self._config.get('allowed_extensions', [])
@property
def max_file_size(self) -> int:
return self._config.get('max_file_size', 10485760)
@property
def max_files_listing(self) -> int:
return self._config.get('max_files_listing', 1000)
@property
def max_recursion_depth(self) -> int:
return self._config.get('max_recursion_depth', 10)
# Global settings instance
settings = Settings()
# Global paths config instance
paths_config_file = settings.config_dir / settings.allowed_paths_config
paths_config = PathsConfig(paths_config_file)

152
src/jetson_agent/main.py Normal file
View File

@ -0,0 +1,152 @@
"""
Jetson Agent API - Main application entry point
FastAPI application for secure file and system access
"""
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
import logging
from pathlib import Path
from .config import settings
from .api import filesystem
# ============================================================================
# Logging Configuration
# ============================================================================
logging.basicConfig(
level=getattr(logging, settings.log_level.upper()),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# ============================================================================
# Lifespan Events
# ============================================================================
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Lifespan context manager for startup/shutdown events
"""
# Startup
logger.info(f"Starting {settings.app_name} v{settings.app_version}")
logger.info(f"Environment: {settings.app_env}")
logger.info(f"Log level: {settings.log_level}")
# Créer le dossier de logs si nécessaire
if settings.log_file:
log_dir = Path(settings.log_file).parent
log_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Logs directory: {log_dir}")
yield
# Shutdown
logger.info(f"Shutting down {settings.app_name}")
# ============================================================================
# FastAPI Application
# ============================================================================
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="Secure API middleware for Jetson Nano file and system access",
lifespan=lifespan,
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json"
)
# ============================================================================
# CORS Middleware
# ============================================================================
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins_list,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["*"],
)
# ============================================================================
# Exception Handlers
# ============================================================================
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""Global exception handler"""
logger.error(f"Unhandled exception: {exc}", exc_info=True)
return JSONResponse(
status_code=500,
content={
"error": "internal_server_error",
"message": "An unexpected error occurred",
"detail": str(exc) if settings.app_env == "development" else None
}
)
# ============================================================================
# Include Routers
# ============================================================================
app.include_router(
filesystem.router,
prefix="/api"
)
# ============================================================================
# Root Endpoints
# ============================================================================
@app.get("/")
async def root():
"""Root endpoint - API information"""
return {
"name": settings.app_name,
"version": settings.app_version,
"environment": settings.app_env,
"status": "operational",
"docs": "/docs",
"endpoints": {
"files": "/api/files/*",
"health": "/api/health"
}
}
@app.get("/api/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"version": settings.app_version,
"environment": settings.app_env
}
# ============================================================================
# Main entry point (for direct execution)
# ============================================================================
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"jetson_agent.main:app",
host=settings.api_host,
port=settings.api_port,
reload=settings.api_reload,
log_level=settings.log_level.lower()
)

View File

@ -0,0 +1,135 @@
"""
Path validation and security module
Ensures file operations are restricted to allowed paths
"""
from pathlib import Path
from typing import Tuple
from fastapi import HTTPException
from .config import paths_config
class PathValidator:
"""Validates file paths against security rules"""
@staticmethod
def validate_read_path(requested_path: str) -> Path:
"""
Validate that a path is allowed for reading
Args:
requested_path: The path requested by the user
Returns:
Resolved Path object
Raises:
HTTPException: If path is not allowed
"""
try:
# Résoudre le chemin (suit les symlinks)
real_path = Path(requested_path).resolve()
# Vérifier que le chemin existe
if not real_path.exists():
raise HTTPException(
status_code=404,
detail=f"Path not found: {requested_path}"
)
# Vérifier qu'il n'y a pas de traversal (..)
if '..' in str(requested_path):
raise HTTPException(
status_code=403,
detail="Path traversal not allowed"
)
# Vérifier contre la blocklist
for blocked in paths_config.blocked_paths:
if str(real_path).startswith(blocked):
raise HTTPException(
status_code=403,
detail=f"Access to {blocked} is forbidden"
)
# Vérifier contre l'allowlist
allowed = False
for allowed_path in paths_config.allowed_read_paths:
if str(real_path).startswith(allowed_path):
allowed = True
break
if not allowed:
raise HTTPException(
status_code=403,
detail=f"Access denied. Path not in allowed list: {requested_path}"
)
return real_path
except HTTPException:
raise
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Error validating path: {str(e)}"
)
@staticmethod
def validate_file_size(file_path: Path) -> Tuple[bool, int]:
"""
Check if file size is within limits
Args:
file_path: Path to the file
Returns:
Tuple of (is_valid, size_in_bytes)
"""
try:
size = file_path.stat().st_size
max_size = paths_config.max_file_size
return (size <= max_size, size)
except Exception:
return (False, 0)
@staticmethod
def validate_file_extension(file_path: Path) -> bool:
"""
Check if file extension is allowed
Args:
file_path: Path to the file
Returns:
True if extension is allowed
"""
extension = file_path.suffix.lower()
# Fichiers sans extension sont autorisés
if not extension:
return True
return extension in paths_config.allowed_extensions
@staticmethod
def is_text_file(file_path: Path) -> bool:
"""
Determine if a file is likely a text file
Args:
file_path: Path to the file
Returns:
True if file appears to be text
"""
text_extensions = {
'.txt', '.md', '.php', '.html', '.css', '.js', '.json',
'.xml', '.yaml', '.yml', '.py', '.sh', '.conf', '.sql',
'.log', '.ini', '.csv', '.tsv'
}
return file_path.suffix.lower() in text_extensions
# Global validator instance
path_validator = PathValidator()