feat: explorador de archivos local y autenticacion/conexion a unidad de red UNC en storage root

This commit is contained in:
Carlos Tello
2026-08-17 00:05:15 -03:00
parent 7badc34816
commit f3b6d8ef71
5 changed files with 457 additions and 25 deletions
+60 -15
View File
@@ -16,8 +16,40 @@ DEFAULT_SETTINGS = {
"radius_host": "",
"radius_port": "1812",
"radius_secret": "",
"storage_network_enabled": "false",
"storage_network_user": "",
"storage_network_pass": "",
}
def connect_network_share(path: str, username: str, password: str) -> bool:
"""
Mounts a UNC network share on Windows using 'net use'.
"""
if not (path.startswith(r"\\") or path.startswith("//")):
return True
parts = [p for p in path.replace("/", "\\").split("\\") if p]
if len(parts) < 2:
return False
unc_base = f"\\\\{parts[0]}\\{parts[1]}"
import subprocess
print(f"Connecting to network share {unc_base} with user '{username}'...")
# 1. Remove existing connection if any (ignore errors)
subprocess.run(["net", "use", unc_base, "/delete", "/y"], capture_output=True)
# 2. Add connection
cmd = ["net", "use", unc_base]
if password:
cmd.append(password)
if username:
cmd.append(f"/user:{username}")
cmd.append("/persistent:yes")
result = subprocess.run(cmd, capture_output=True, text=True)
return result.returncode == 0
async def get_setting(db: AsyncSession, key: str) -> str:
result = await db.execute(select(SystemSetting).where(SystemSetting.key == key))
setting = result.scalar_one_or_none()
@@ -43,23 +75,36 @@ async def get_all_settings(db: AsyncSession) -> dict:
async def update_settings_service(db: AsyncSession, new_settings: dict) -> dict:
from app.storage.local import storage_provider
current = await get_all_settings(db)
merged = {**current, **new_settings}
# Connect network share if enabled and it's a UNC path
net_enabled = str(merged.get("storage_network_enabled", "false")).lower() == "true"
net_user = merged.get("storage_network_user", "")
net_pass = merged.get("storage_network_pass", "")
root_path = merged.get("storage_root", "")
if net_enabled and root_path and (root_path.startswith(r"\\") or root_path.startswith("//")):
ok = connect_network_share(root_path, net_user, net_pass)
if not ok:
raise ValueError("Failed to authenticate or connect to the specified network share (UNC). Verify path and credentials.")
# Perform standard storage root validation
if "storage_root" in new_settings:
v = new_settings["storage_root"]
path = Path(v).resolve()
try:
os.makedirs(path, exist_ok=True)
test_file = path / ".write_test"
test_file.touch()
test_file.unlink()
except Exception as e:
raise ValueError(f"Invalid or unwritable storage root path: {str(e)}")
storage_provider.root_dir = path
for k, v in new_settings.items():
if k in DEFAULT_SETTINGS and v is not None:
# If changing storage root, validate and apply
if k == "storage_root":
path = Path(v).resolve()
try:
os.makedirs(path, exist_ok=True)
# Test write access
test_file = path / ".write_test"
test_file.touch()
test_file.unlink()
except Exception as e:
raise ValueError(f"Invalid or unwritable storage root path: {str(e)}")
# Apply to in-memory storage provider
storage_provider.root_dir = path
result = await db.execute(select(SystemSetting).where(SystemSetting.key == k))
setting = result.scalar_one_or_none()
if setting: