from datetime import datetime, timedelta, timezone
from typing import Any
import jwt
import psycopg
import redis
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
from fastapi import Depends, FastAPI, Header, HTTPException, status
from pydantic import BaseModel, EmailStr, Field
from app.config import settings
app = FastAPI(title=settings.app_name)
password_hasher = PasswordHasher()
def api_response(data: Any = None, meta: dict | None = None, error: Any = None) -> dict:
return {
"data": data,
"meta": meta or {},
"error": error,
}
def json_safe(value: Any) -> Any:
if isinstance(value, dict):
return {key: json_safe(item) for key, item in value.items()}
if isinstance(value, list):
return [json_safe(item) for item in value]
if isinstance(value, datetime):
return value.isoformat()
return value
def get_db_connection():
return psycopg.connect(
host=settings.postgres_host,
port=settings.postgres_port,
dbname=settings.postgres_db,
user=settings.postgres_user,
password=settings.postgres_password,
connect_timeout=5,
)
def check_database() -> str:
conn = None
try:
conn = get_db_connection()
with conn.cursor() as cur:
cur.execute("SELECT 1;")
result = cur.fetchone()
if result and result[0] == 1:
return "ok"
return "error"
except Exception as exc:
return f"error: {exc}"
finally:
if conn is not None:
conn.close()
def check_redis() -> str:
try:
client = redis.Redis.from_url(settings.redis_url, socket_connect_timeout=5)
if client.ping():
return "ok"
return "error"
except Exception as exc:
return f"error: {exc}"
class SetupAdminRequest(BaseModel):
username: str = Field(min_length=3, max_length=64)
display_name: str = Field(min_length=1, max_length=120)
email: EmailStr | None = None
password: str = Field(min_length=10, max_length=256)
class LoginRequest(BaseModel):
username: str
password: str
def create_access_token(user_id: str, username: str, role: str) -> str:
now = datetime.now(timezone.utc)
payload = {
"sub": user_id,
"username": username,
"role": role,
"iat": int(now.timestamp()),
"exp": int((now + timedelta(minutes=settings.jwt_access_token_minutes)).timestamp()),
"type": "access",
}
return jwt.encode(payload, settings.jwt_secret_key, algorithm="HS256")
def decode_access_token(token: str) -> dict:
try:
payload = jwt.decode(token, settings.jwt_secret_key, algorithms=["HS256"])
if payload.get("type") != "access":
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type")
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired")
except jwt.InvalidTokenError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
def get_bearer_token(authorization: str | None = Header(default=None)) -> str:
if not authorization or not authorization.lower().startswith("bearer "):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
return authorization.split(" ", 1)[1].strip()
def get_current_user(token: str = Depends(get_bearer_token)) -> dict:
payload = decode_access_token(token)
user_id = payload["sub"]
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
users.id::text AS id,
users.username,
users.display_name,
users.email,
users.is_active,
roles.name AS role
FROM users
JOIN roles ON roles.id = users.role_id
WHERE users.id = %s;
""",
(user_id,),
)
user = cur.fetchone()
if not user or not user["is_active"]:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive")
return dict(user)
finally:
if conn is not None:
conn.close()
@app.get("/api/health")
def health_check():
database_status = check_database()
redis_status = check_redis()
overall_status = "ok"
if database_status != "ok" or redis_status != "ok":
overall_status = "degraded"
return api_response(
data={
"status": overall_status,
"app": settings.app_name,
"environment": settings.app_env,
"database": database_status,
"redis": redis_status,
}
)
@app.get("/api/auth/setup-status")
def setup_status():
conn = None
try:
conn = get_db_connection()
with conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM users;")
user_count = cur.fetchone()[0]
return api_response(
data={
"setup_required": user_count == 0,
"user_count": user_count,
}
)
finally:
if conn is not None:
conn.close()
@app.post("/api/auth/setup-admin")
def setup_admin(request: SetupAdminRequest):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute("SELECT COUNT(*) AS user_count FROM users;")
user_count = cur.fetchone()["user_count"]
if user_count > 0:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Initial admin setup has already been completed",
)
cur.execute("SELECT id FROM roles WHERE name = 'Admin';")
admin_role = cur.fetchone()
if not admin_role:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Admin role is missing",
)
password_hash = password_hasher.hash(request.password)
cur.execute(
"""
INSERT INTO users (username, display_name, email, password_hash, role_id)
VALUES (%s, %s, %s, %s, %s)
RETURNING id::text, username, display_name, email;
""",
(
request.username.strip(),
request.display_name.strip(),
request.email,
password_hash,
admin_role["id"],
),
)
user = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'user.created', 'user', %s, %s);
""",
(
user["id"],
user["id"],
psycopg.types.json.Jsonb(
{
"username": user["username"],
"display_name": user["display_name"],
"role": "Admin",
"source": "setup-admin",
}
),
),
)
conn.commit()
return api_response(
data={
"user_id": user["id"],
"username": user["username"],
"display_name": user["display_name"],
}
)
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.post("/api/auth/login")
def login(request: LoginRequest):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
users.id::text AS id,
users.username,
users.display_name,
users.email,
users.password_hash,
users.is_active,
roles.name AS role
FROM users
JOIN roles ON roles.id = users.role_id
WHERE lower(users.username) = lower(%s);
""",
(request.username.strip(),),
)
user = cur.fetchone()
if not user or not user["is_active"]:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid username or password")
try:
password_hasher.verify(user["password_hash"], request.password)
except VerifyMismatchError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid username or password")
access_token = create_access_token(user["id"], user["username"], user["role"])
cur.execute(
"UPDATE users SET last_login_at = now(), updated_at = now() WHERE id = %s;",
(user["id"],),
)
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'user.login', 'user', %s, %s);
""",
(
user["id"],
user["id"],
psycopg.types.json.Jsonb({"username": user["username"]}),
),
)
conn.commit()
return api_response(
data={
"access_token": access_token,
"token_type": "bearer",
"expires_in": settings.jwt_access_token_minutes * 60,
"user": {
"id": user["id"],
"username": user["username"],
"display_name": user["display_name"],
"email": user["email"],
"role": user["role"],
},
}
)
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.get("/api/auth/me")
def me(current_user: dict = Depends(get_current_user)):
return api_response(data=current_user)
class CreateRoomRequest(BaseModel):
site_id: str | None = None
name: str = Field(min_length=1, max_length=120)
description: str | None = None
sort_order: int = 0
class UpdateRoomRequest(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=120)
description: str | None = None
sort_order: int | None = None
@app.get("/api/sites")
def list_sites(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT id::text, name, description, created_at, updated_at
FROM sites
ORDER BY name;
"""
)
sites = [dict(row) for row in cur.fetchall()]
return api_response(data=sites)
finally:
if conn is not None:
conn.close()
@app.get("/api/rooms")
def list_rooms(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
rooms.id::text,
rooms.site_id::text,
sites.name AS site_name,
rooms.name,
rooms.description,
rooms.sort_order,
rooms.created_at,
rooms.updated_at
FROM rooms
JOIN sites ON sites.id = rooms.site_id
ORDER BY rooms.sort_order, rooms.name;
"""
)
rooms = [dict(row) for row in cur.fetchall()]
return api_response(data=rooms)
finally:
if conn is not None:
conn.close()
@app.post("/api/rooms")
def create_room(request: CreateRoomRequest, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
site_id = request.site_id
if site_id is None:
cur.execute("SELECT id::text FROM sites ORDER BY name LIMIT 1;")
site = cur.fetchone()
if not site:
raise HTTPException(status_code=500, detail="No site exists")
site_id = site["id"]
cur.execute(
"""
INSERT INTO rooms (site_id, name, description, sort_order)
VALUES (%s, %s, %s, %s)
RETURNING id::text, site_id::text, name, description, sort_order, created_at, updated_at;
""",
(
site_id,
request.name.strip(),
request.description,
request.sort_order,
),
)
room = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'room.created', 'room', %s, %s);
""",
(
current_user["id"],
room["id"],
psycopg.types.json.Jsonb(
{
"name": room["name"],
"site_id": room["site_id"],
"sort_order": room["sort_order"],
}
),
),
)
conn.commit()
return api_response(data=dict(room))
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.patch("/api/rooms/{room_id}")
def update_room(room_id: str, request: UpdateRoomRequest, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT id::text, name, description, sort_order
FROM rooms
WHERE id = %s;
""",
(room_id,),
)
old_room = cur.fetchone()
if not old_room:
raise HTTPException(status_code=404, detail="Room not found")
new_name = request.name.strip() if request.name is not None else old_room["name"]
new_description = request.description if request.description is not None else old_room["description"]
new_sort_order = request.sort_order if request.sort_order is not None else old_room["sort_order"]
cur.execute(
"""
UPDATE rooms
SET name = %s,
description = %s,
sort_order = %s,
updated_at = now()
WHERE id = %s
RETURNING id::text, site_id::text, name, description, sort_order, created_at, updated_at;
""",
(new_name, new_description, new_sort_order, room_id),
)
room = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, old_value_json, new_value_json)
VALUES (%s, 'room.updated', 'room', %s, %s, %s);
""",
(
current_user["id"],
room["id"],
psycopg.types.json.Jsonb(json_safe(dict(old_room))),
psycopg.types.json.Jsonb(json_safe(dict(room))),
),
)
conn.commit()
return api_response(data=dict(room))
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.delete("/api/rooms/{room_id}")
def delete_room(room_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT id::text, name, description, sort_order
FROM rooms
WHERE id = %s;
""",
(room_id,),
)
room = cur.fetchone()
if not room:
raise HTTPException(status_code=404, detail="Room not found")
cur.execute("DELETE FROM rooms WHERE id = %s;", (room_id,))
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, old_value_json)
VALUES (%s, 'room.deleted', 'room', %s, %s);
""",
(
current_user["id"],
room["id"],
psycopg.types.json.Jsonb(json_safe(dict(room))),
),
)
conn.commit()
return api_response(data={"deleted": True, "room_id": room_id})
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
class CreateRoomRequest(BaseModel):
site_id: str | None = None
name: str = Field(min_length=1, max_length=120)
description: str | None = None
sort_order: int = 0
class UpdateRoomRequest(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=120)
description: str | None = None
sort_order: int | None = None
@app.get("/api/sites")
def list_sites(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT id::text, name, description, created_at, updated_at
FROM sites
ORDER BY name;
"""
)
sites = [dict(row) for row in cur.fetchall()]
return api_response(data=sites)
finally:
if conn is not None:
conn.close()
@app.get("/api/rooms")
def list_rooms(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
rooms.id::text,
rooms.site_id::text,
sites.name AS site_name,
rooms.name,
rooms.description,
rooms.sort_order,
rooms.created_at,
rooms.updated_at
FROM rooms
JOIN sites ON sites.id = rooms.site_id
ORDER BY rooms.sort_order, rooms.name;
"""
)
rooms = [dict(row) for row in cur.fetchall()]
return api_response(data=rooms)
finally:
if conn is not None:
conn.close()
@app.post("/api/rooms")
def create_room(request: CreateRoomRequest, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
site_id = request.site_id
if site_id is None:
cur.execute("SELECT id::text FROM sites ORDER BY name LIMIT 1;")
site = cur.fetchone()
if not site:
raise HTTPException(status_code=500, detail="No site exists")
site_id = site["id"]
cur.execute(
"""
INSERT INTO rooms (site_id, name, description, sort_order)
VALUES (%s, %s, %s, %s)
RETURNING id::text, site_id::text, name, description, sort_order, created_at, updated_at;
""",
(
site_id,
request.name.strip(),
request.description,
request.sort_order,
),
)
room = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'room.created', 'room', %s, %s);
""",
(
current_user["id"],
room["id"],
psycopg.types.json.Jsonb(
{
"name": room["name"],
"site_id": room["site_id"],
"sort_order": room["sort_order"],
}
),
),
)
conn.commit()
return api_response(data=dict(room))
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.patch("/api/rooms/{room_id}")
def update_room(room_id: str, request: UpdateRoomRequest, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT id::text, name, description, sort_order
FROM rooms
WHERE id = %s;
""",
(room_id,),
)
old_room = cur.fetchone()
if not old_room:
raise HTTPException(status_code=404, detail="Room not found")
new_name = request.name.strip() if request.name is not None else old_room["name"]
new_description = request.description if request.description is not None else old_room["description"]
new_sort_order = request.sort_order if request.sort_order is not None else old_room["sort_order"]
cur.execute(
"""
UPDATE rooms
SET name = %s,
description = %s,
sort_order = %s,
updated_at = now()
WHERE id = %s
RETURNING id::text, site_id::text, name, description, sort_order, created_at, updated_at;
""",
(new_name, new_description, new_sort_order, room_id),
)
room = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, old_value_json, new_value_json)
VALUES (%s, 'room.updated', 'room', %s, %s, %s);
""",
(
current_user["id"],
room["id"],
psycopg.types.json.Jsonb(json_safe(dict(old_room))),
psycopg.types.json.Jsonb(json_safe(dict(room))),
),
)
conn.commit()
return api_response(data=dict(room))
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.delete("/api/rooms/{room_id}")
def delete_room(room_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT id::text, name, description, sort_order
FROM rooms
WHERE id = %s;
""",
(room_id,),
)
room = cur.fetchone()
if not room:
raise HTTPException(status_code=404, detail="Room not found")
cur.execute("DELETE FROM rooms WHERE id = %s;", (room_id,))
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, old_value_json)
VALUES (%s, 'room.deleted', 'room', %s, %s);
""",
(
current_user["id"],
room["id"],
psycopg.types.json.Jsonb(json_safe(dict(room))),
),
)
conn.commit()
return api_response(data={"deleted": True, "room_id": room_id})
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
import ipaddress
class CreateScanRangeRequest(BaseModel):
room_id: str | None = None
name: str = Field(min_length=1, max_length=120)
range_type: str = "cidr"
cidr: str | None = None
start_ip: str | None = None
end_ip: str | None = None
enabled: bool = True
scan_interval_seconds: int = Field(default=900, ge=60, le=86400)
timeout_ms: int = Field(default=3000, ge=500, le=30000)
retry_limit: int = Field(default=1, ge=0, le=5)
parallelism_limit: int = Field(default=10, ge=1, le=100)
stagger_ms: int = Field(default=250, ge=0, le=10000)
api_port: int = Field(default=4028, ge=1, le=65535)
class UpdateScanRangeRequest(BaseModel):
room_id: str | None = None
name: str | None = Field(default=None, min_length=1, max_length=120)
range_type: str | None = None
cidr: str | None = None
start_ip: str | None = None
end_ip: str | None = None
enabled: bool | None = None
scan_interval_seconds: int | None = Field(default=None, ge=60, le=86400)
timeout_ms: int | None = Field(default=None, ge=500, le=30000)
retry_limit: int | None = Field(default=None, ge=0, le=5)
parallelism_limit: int | None = Field(default=None, ge=1, le=100)
stagger_ms: int | None = Field(default=None, ge=0, le=10000)
api_port: int | None = Field(default=None, ge=1, le=65535)
def validate_scan_range_input(range_type: str, cidr: str | None, start_ip: str | None, end_ip: str | None) -> None:
if range_type not in ("cidr", "start_end"):
raise HTTPException(status_code=422, detail="range_type must be 'cidr' or 'start_end'")
if range_type == "cidr":
if not cidr:
raise HTTPException(status_code=422, detail="cidr is required when range_type is cidr")
try:
ipaddress.ip_network(cidr, strict=False)
except ValueError as exc:
raise HTTPException(status_code=422, detail=f"Invalid CIDR range: {exc}")
if range_type == "start_end":
if not start_ip or not end_ip:
raise HTTPException(status_code=422, detail="start_ip and end_ip are required when range_type is start_end")
try:
start = ipaddress.ip_address(start_ip)
end = ipaddress.ip_address(end_ip)
except ValueError as exc:
raise HTTPException(status_code=422, detail=f"Invalid IP address: {exc}")
if start.version != end.version:
raise HTTPException(status_code=422, detail="start_ip and end_ip must use the same IP version")
if int(start) > int(end):
raise HTTPException(status_code=422, detail="start_ip must be lower than or equal to end_ip")
@app.get("/api/scan-ranges")
def list_scan_ranges(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
scan_ranges.id::text,
scan_ranges.room_id::text,
rooms.name AS room_name,
scan_ranges.name,
scan_ranges.range_type,
scan_ranges.cidr,
host(scan_ranges.start_ip) AS start_ip,
host(scan_ranges.end_ip) AS end_ip,
scan_ranges.enabled,
scan_ranges.scan_interval_seconds,
scan_ranges.timeout_ms,
scan_ranges.retry_limit,
scan_ranges.parallelism_limit,
scan_ranges.stagger_ms,
scan_ranges.api_port,
scan_ranges.last_scan_at,
scan_ranges.next_scan_at,
scan_ranges.created_at,
scan_ranges.updated_at
FROM scan_ranges
LEFT JOIN rooms ON rooms.id = scan_ranges.room_id
ORDER BY rooms.sort_order, scan_ranges.name;
"""
)
ranges = [dict(row) for row in cur.fetchall()]
return api_response(data=ranges)
finally:
if conn is not None:
conn.close()
@app.post("/api/scan-ranges")
def create_scan_range(request: CreateScanRangeRequest, current_user: dict = Depends(get_current_user)):
validate_scan_range_input(request.range_type, request.cidr, request.start_ip, request.end_ip)
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
INSERT INTO scan_ranges (
room_id,
name,
range_type,
cidr,
start_ip,
end_ip,
enabled,
scan_interval_seconds,
timeout_ms,
retry_limit,
parallelism_limit,
stagger_ms,
api_port,
created_by_user_id,
next_scan_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now())
RETURNING
id::text,
room_id::text,
name,
range_type,
cidr,
host(start_ip) AS start_ip,
host(end_ip) AS end_ip,
enabled,
scan_interval_seconds,
timeout_ms,
retry_limit,
parallelism_limit,
stagger_ms,
api_port,
last_scan_at,
next_scan_at,
created_at,
updated_at;
""",
(
request.room_id,
request.name.strip(),
request.range_type,
request.cidr,
request.start_ip,
request.end_ip,
request.enabled,
request.scan_interval_seconds,
request.timeout_ms,
request.retry_limit,
request.parallelism_limit,
request.stagger_ms,
request.api_port,
current_user["id"],
),
)
scan_range = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'scan_range.created', 'scan_range', %s, %s);
""",
(
current_user["id"],
scan_range["id"],
psycopg.types.json.Jsonb(json_safe(dict(scan_range))),
),
)
conn.commit()
return api_response(data=dict(scan_range))
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.patch("/api/scan-ranges/{scan_range_id}")
def update_scan_range(scan_range_id: str, request: UpdateScanRangeRequest, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
room_id::text,
name,
range_type,
cidr,
host(start_ip) AS start_ip,
host(end_ip) AS end_ip,
enabled,
scan_interval_seconds,
timeout_ms,
retry_limit,
parallelism_limit,
stagger_ms,
api_port
FROM scan_ranges
WHERE id = %s;
""",
(scan_range_id,),
)
old_range = cur.fetchone()
if not old_range:
raise HTTPException(status_code=404, detail="Scan range not found")
new_room_id = request.room_id if request.room_id is not None else old_range["room_id"]
new_name = request.name.strip() if request.name is not None else old_range["name"]
new_range_type = request.range_type if request.range_type is not None else old_range["range_type"]
new_cidr = request.cidr if request.cidr is not None else old_range["cidr"]
new_start_ip = request.start_ip if request.start_ip is not None else old_range["start_ip"]
new_end_ip = request.end_ip if request.end_ip is not None else old_range["end_ip"]
new_enabled = request.enabled if request.enabled is not None else old_range["enabled"]
new_scan_interval_seconds = request.scan_interval_seconds if request.scan_interval_seconds is not None else old_range["scan_interval_seconds"]
new_timeout_ms = request.timeout_ms if request.timeout_ms is not None else old_range["timeout_ms"]
new_retry_limit = request.retry_limit if request.retry_limit is not None else old_range["retry_limit"]
new_parallelism_limit = request.parallelism_limit if request.parallelism_limit is not None else old_range["parallelism_limit"]
new_stagger_ms = request.stagger_ms if request.stagger_ms is not None else old_range["stagger_ms"]
new_api_port = request.api_port if request.api_port is not None else old_range["api_port"]
validate_scan_range_input(new_range_type, new_cidr, new_start_ip, new_end_ip)
cur.execute(
"""
UPDATE scan_ranges
SET room_id = %s,
name = %s,
range_type = %s,
cidr = %s,
start_ip = %s,
end_ip = %s,
enabled = %s,
scan_interval_seconds = %s,
timeout_ms = %s,
retry_limit = %s,
parallelism_limit = %s,
stagger_ms = %s,
api_port = %s,
updated_at = now()
WHERE id = %s
RETURNING
id::text,
room_id::text,
name,
range_type,
cidr,
host(start_ip) AS start_ip,
host(end_ip) AS end_ip,
enabled,
scan_interval_seconds,
timeout_ms,
retry_limit,
parallelism_limit,
stagger_ms,
api_port,
last_scan_at,
next_scan_at,
created_at,
updated_at;
""",
(
new_room_id,
new_name,
new_range_type,
new_cidr,
new_start_ip,
new_end_ip,
new_enabled,
new_scan_interval_seconds,
new_timeout_ms,
new_retry_limit,
new_parallelism_limit,
new_stagger_ms,
new_api_port,
scan_range_id,
),
)
scan_range = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, old_value_json, new_value_json)
VALUES (%s, 'scan_range.updated', 'scan_range', %s, %s, %s);
""",
(
current_user["id"],
scan_range["id"],
psycopg.types.json.Jsonb(json_safe(dict(old_range))),
psycopg.types.json.Jsonb(json_safe(dict(scan_range))),
),
)
conn.commit()
return api_response(data=dict(scan_range))
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.delete("/api/scan-ranges/{scan_range_id}")
def delete_scan_range(scan_range_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT id::text, name, range_type, cidr, host(start_ip) AS start_ip, end_ip::text
FROM scan_ranges
WHERE id = %s;
""",
(scan_range_id,),
)
old_range = cur.fetchone()
if not old_range:
raise HTTPException(status_code=404, detail="Scan range not found")
cur.execute("DELETE FROM scan_ranges WHERE id = %s;", (scan_range_id,))
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, old_value_json)
VALUES (%s, 'scan_range.deleted', 'scan_range', %s, %s);
""",
(
current_user["id"],
old_range["id"],
psycopg.types.json.Jsonb(json_safe(dict(old_range))),
),
)
conn.commit()
return api_response(data={"deleted": True, "scan_range_id": scan_range_id})
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.get("/api/scan-runs")
def list_scan_runs(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
scan_runs.id::text,
scan_runs.scan_range_id::text,
scan_ranges.name AS scan_range_name,
rooms.name AS room_name,
scan_runs.triggered_by,
scan_runs.triggered_by_user_id::text,
scan_runs.started_at,
scan_runs.finished_at,
scan_runs.status,
scan_runs.total_ips,
scan_runs.scanned_ips,
scan_runs.miners_found,
scan_runs.errors_count,
scan_runs.notes
FROM scan_runs
LEFT JOIN scan_ranges ON scan_ranges.id = scan_runs.scan_range_id
LEFT JOIN rooms ON rooms.id = scan_ranges.room_id
ORDER BY scan_runs.started_at DESC
LIMIT 100;
"""
)
scan_runs = [dict(row) for row in cur.fetchall()]
return api_response(data=scan_runs)
finally:
if conn is not None:
conn.close()
@app.get("/api/scan-runs/{scan_run_id}")
def get_scan_run(scan_run_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
scan_runs.id::text,
scan_runs.scan_range_id::text,
scan_ranges.name AS scan_range_name,
rooms.name AS room_name,
scan_runs.triggered_by,
scan_runs.triggered_by_user_id::text,
scan_runs.started_at,
scan_runs.finished_at,
scan_runs.status,
scan_runs.total_ips,
scan_runs.scanned_ips,
scan_runs.miners_found,
scan_runs.errors_count,
scan_runs.notes
FROM scan_runs
LEFT JOIN scan_ranges ON scan_ranges.id = scan_runs.scan_range_id
LEFT JOIN rooms ON rooms.id = scan_ranges.room_id
WHERE scan_runs.id = %s;
""",
(scan_run_id,),
)
scan_run = cur.fetchone()
if not scan_run:
raise HTTPException(status_code=404, detail="Scan run not found")
return api_response(data=dict(scan_run))
finally:
if conn is not None:
conn.close()
@app.get("/api/scan-runs/{scan_run_id}/results")
def list_scan_run_results(scan_run_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
scan_run_id::text,
host(ip_address) AS ip_address,
port,
detected,
mac_address,
response_ms,
status,
error_message,
raw_summary_json,
raw_version_json,
raw_pools_json,
raw_error_json,
created_at
FROM scan_results
WHERE scan_run_id = %s
ORDER BY ip_address
LIMIT 2000;
""",
(scan_run_id,),
)
results = [dict(row) for row in cur.fetchall()]
return api_response(data=results)
finally:
if conn is not None:
conn.close()
@app.get("/api/scan-runs")
def list_scan_runs(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
scan_runs.id::text,
scan_runs.scan_range_id::text,
scan_ranges.name AS scan_range_name,
rooms.name AS room_name,
scan_runs.triggered_by,
scan_runs.triggered_by_user_id::text,
scan_runs.started_at,
scan_runs.finished_at,
scan_runs.status,
scan_runs.total_ips,
scan_runs.scanned_ips,
scan_runs.miners_found,
scan_runs.errors_count,
scan_runs.notes
FROM scan_runs
LEFT JOIN scan_ranges ON scan_ranges.id = scan_runs.scan_range_id
LEFT JOIN rooms ON rooms.id = scan_ranges.room_id
ORDER BY scan_runs.started_at DESC
LIMIT 100;
"""
)
scan_runs = [dict(row) for row in cur.fetchall()]
return api_response(data=scan_runs)
finally:
if conn is not None:
conn.close()
@app.get("/api/scan-runs/{scan_run_id}")
def get_scan_run(scan_run_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
scan_runs.id::text,
scan_runs.scan_range_id::text,
scan_ranges.name AS scan_range_name,
rooms.name AS room_name,
scan_runs.triggered_by,
scan_runs.triggered_by_user_id::text,
scan_runs.started_at,
scan_runs.finished_at,
scan_runs.status,
scan_runs.total_ips,
scan_runs.scanned_ips,
scan_runs.miners_found,
scan_runs.errors_count,
scan_runs.notes
FROM scan_runs
LEFT JOIN scan_ranges ON scan_ranges.id = scan_runs.scan_range_id
LEFT JOIN rooms ON rooms.id = scan_ranges.room_id
WHERE scan_runs.id = %s;
""",
(scan_run_id,),
)
scan_run = cur.fetchone()
if not scan_run:
raise HTTPException(status_code=404, detail="Scan run not found")
return api_response(data=dict(scan_run))
finally:
if conn is not None:
conn.close()
@app.get("/api/scan-runs/{scan_run_id}/results")
def list_scan_run_results(scan_run_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
scan_run_id::text,
host(ip_address) AS ip_address,
port,
detected,
mac_address,
response_ms,
status,
error_message,
raw_summary_json,
raw_version_json,
raw_pools_json,
raw_error_json,
created_at
FROM scan_results
WHERE scan_run_id = %s
ORDER BY ip_address
LIMIT 2000;
""",
(scan_run_id,),
)
results = [dict(row) for row in cur.fetchall()]
return api_response(data=results)
finally:
if conn is not None:
conn.close()
import socket
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
class ScanNowRequest(BaseModel):
max_ips: int = Field(default=25, ge=1, le=2000)
specific_ips: list[str] | None = None
def expand_scan_range_for_scan(scan_range: dict, max_ips: int, specific_ips: list[str] | None = None) -> list[str]:
if specific_ips:
validated_ips = []
for ip in specific_ips:
try:
validated_ips.append(str(ipaddress.ip_address(ip)))
except ValueError as exc:
raise HTTPException(status_code=422, detail=f"Invalid IP address {ip}: {exc}")
return validated_ips[:max_ips]
range_type = scan_range["range_type"]
if range_type == "cidr":
network = ipaddress.ip_network(scan_range["cidr"], strict=False)
return [str(ip) for ip in list(network.hosts())[:max_ips]]
if range_type == "start_end":
start = ipaddress.ip_address(scan_range["start_ip"])
end = ipaddress.ip_address(scan_range["end_ip"])
ips = []
current = int(start)
end_int = int(end)
while current <= end_int and len(ips) < max_ips:
ips.append(str(ipaddress.ip_address(current)))
current += 1
return ips
raise HTTPException(status_code=422, detail="Unsupported scan range type")
def tcp_check_ip(ip_address: str, port: int, timeout_ms: int) -> dict:
started = time.perf_counter()
try:
with socket.create_connection((ip_address, port), timeout=timeout_ms / 1000):
elapsed_ms = int((time.perf_counter() - started) * 1000)
return {
"ip_address": ip_address,
"port": port,
"detected": True,
"response_ms": elapsed_ms,
"status": "port_open",
"error_message": None,
}
except socket.timeout:
elapsed_ms = int((time.perf_counter() - started) * 1000)
return {
"ip_address": ip_address,
"port": port,
"detected": False,
"response_ms": elapsed_ms,
"status": "timeout",
"error_message": "Connection timed out",
}
except OSError as exc:
elapsed_ms = int((time.perf_counter() - started) * 1000)
return {
"ip_address": ip_address,
"port": port,
"detected": False,
"response_ms": elapsed_ms,
"status": "no_response",
"error_message": str(exc),
}
@app.post("/api/scan-ranges/{scan_range_id}/scan-now")
def scan_range_now(scan_range_id: str, request: ScanNowRequest, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
room_id::text,
name,
range_type,
cidr,
host(start_ip) AS start_ip,
host(end_ip) AS end_ip,
enabled,
timeout_ms,
retry_limit,
parallelism_limit,
stagger_ms,
api_port,
scan_interval_seconds
FROM scan_ranges
WHERE id = %s;
""",
(scan_range_id,),
)
scan_range = cur.fetchone()
if not scan_range:
raise HTTPException(status_code=404, detail="Scan range not found")
if not scan_range["enabled"]:
raise HTTPException(status_code=409, detail="Scan range is disabled")
ip_list = expand_scan_range_for_scan(dict(scan_range), request.max_ips, request.specific_ips)
cur.execute(
"""
INSERT INTO scan_runs (
scan_range_id,
triggered_by,
triggered_by_user_id,
status,
total_ips,
scanned_ips,
miners_found,
errors_count,
notes
)
VALUES (%s, 'manual', %s, 'running', %s, 0, 0, 0, %s)
RETURNING id::text;
""",
(
scan_range_id,
current_user["id"],
len(ip_list),
"TCP port 4028 reachability scan only",
),
)
scan_run_id = cur.fetchone()["id"]
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'scan.started', 'scan_run', %s, %s);
""",
(
current_user["id"],
scan_run_id,
psycopg.types.json.Jsonb(
{
"scan_range_id": scan_range_id,
"scan_range_name": scan_range["name"],
"total_ips": len(ip_list),
"mode": "tcp_4028_only",
}
),
),
)
conn.commit()
results = []
detected_count = 0
errors_count = 0
max_workers = min(scan_range["parallelism_limit"], len(ip_list))
if max_workers < 1:
max_workers = 1
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {
executor.submit(
tcp_check_ip,
ip,
scan_range["api_port"],
scan_range["timeout_ms"],
): ip
for ip in ip_list
}
for future in as_completed(future_map):
result = future.result()
results.append(result)
if result["detected"]:
detected_count += 1
if result["status"] not in ("port_open", "timeout", "no_response"):
errors_count += 1
results = sorted(results, key=lambda item: ipaddress.ip_address(item["ip_address"]))
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
for result in results:
cur.execute(
"""
INSERT INTO scan_results (
scan_run_id,
ip_address,
port,
detected,
response_ms,
status,
error_message
)
VALUES (%s, %s, %s, %s, %s, %s, %s);
""",
(
scan_run_id,
result["ip_address"],
result["port"],
result["detected"],
result["response_ms"],
result["status"],
result["error_message"],
),
)
final_status = "completed"
if errors_count > 0:
final_status = "completed_with_errors"
cur.execute(
"""
UPDATE scan_runs
SET finished_at = now(),
status = %s,
scanned_ips = %s,
miners_found = %s,
errors_count = %s
WHERE id = %s
RETURNING
id::text,
scan_range_id::text,
triggered_by,
started_at,
finished_at,
status,
total_ips,
scanned_ips,
miners_found,
errors_count,
notes;
""",
(
final_status,
len(results),
detected_count,
errors_count,
scan_run_id,
),
)
scan_run = cur.fetchone()
cur.execute(
"""
UPDATE scan_ranges
SET last_scan_at = now(),
next_scan_at = now() + make_interval(secs => scan_interval_seconds),
updated_at = now()
WHERE id = %s;
""",
(scan_range_id,),
)
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'scan.completed', 'scan_run', %s, %s);
""",
(
current_user["id"],
scan_run_id,
psycopg.types.json.Jsonb(
{
"scan_range_id": scan_range_id,
"total_ips": len(ip_list),
"scanned_ips": len(results),
"port_4028_open": detected_count,
"errors_count": errors_count,
}
),
),
)
conn.commit()
return api_response(
data={
"scan_run": dict(scan_run),
"summary": {
"scan_range_id": scan_range_id,
"scan_range_name": scan_range["name"],
"total_ips": len(ip_list),
"scanned_ips": len(results),
"port_4028_open": detected_count,
"errors_count": errors_count,
},
}
)
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
class WhatsminerReadTestRequest(BaseModel):
ip_address: str
port: int = Field(default=4028, ge=1, le=65535)
command: str = Field(default="summary", min_length=1, max_length=80)
params: dict[str, Any] | None = None
timeout_ms: int = Field(default=5000, ge=500, le=30000)
def read_from_socket_until_closed(sock: socket.socket, max_bytes: int = 1048576) -> bytes:
chunks = []
total = 0
while True:
chunk = sock.recv(65536)
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
if total >= max_bytes:
break
return b"".join(chunks)
def send_whatsminer_read_command(ip_address: str, port: int, command: str, timeout_ms: int, params: dict[str, Any] | None = None) -> dict:
try:
ipaddress.ip_address(ip_address)
except ValueError as exc:
raise HTTPException(status_code=422, detail=f"Invalid IP address: {exc}")
import json
started = time.perf_counter()
params = params or {}
command_payload = {"command": command}
command_payload.update(params)
cmd_payload = {"cmd": command}
cmd_payload.update(params)
command_payloads = [
{
"name": "json_command_plain",
"payload": json.dumps(command_payload, separators=(",", ":")).encode("utf-8"),
},
{
"name": "json_command_newline",
"payload": (json.dumps(command_payload, separators=(",", ":")) + "\n").encode("utf-8"),
},
{
"name": "json_cmd_plain",
"payload": json.dumps(cmd_payload, separators=(",", ":")).encode("utf-8"),
},
{
"name": "json_cmd_newline",
"payload": (json.dumps(cmd_payload, separators=(",", ":")) + "\n").encode("utf-8"),
},
{
"name": "plain_command",
"payload": command.encode("utf-8"),
},
{
"name": "plain_command_newline",
"payload": (command + "\n").encode("utf-8"),
},
]
attempts = []
for command_payload in command_payloads:
payload_name = command_payload["name"]
payload = command_payload["payload"]
try:
with socket.create_connection((ip_address, port), timeout=timeout_ms / 1000) as sock:
sock.settimeout(timeout_ms / 1000)
sock.sendall(payload)
raw_response = read_from_socket_until_closed(sock)
elapsed_ms = int((time.perf_counter() - started) * 1000)
response_text = raw_response.decode("utf-8", errors="replace").strip()
attempts.append(
{
"payload_type": payload_name,
"bytes_received": len(raw_response),
"response_preview": response_text[:1000],
}
)
if response_text:
parsed_json = None
parse_error = None
try:
parsed_json = json.loads(response_text)
except Exception as exc:
parse_error = str(exc)
return {
"success": True,
"ip_address": ip_address,
"port": port,
"command": command,
"payload_type": payload_name,
"response_ms": elapsed_ms,
"bytes_received": len(raw_response),
"response_text": response_text,
"parsed_json": parsed_json,
"parse_error": parse_error,
"attempts": attempts,
}
except Exception as exc:
attempts.append(
{
"payload_type": payload_name,
"bytes_received": 0,
"error": str(exc),
}
)
elapsed_ms = int((time.perf_counter() - started) * 1000)
return {
"success": False,
"ip_address": ip_address,
"port": port,
"command": command,
"response_ms": elapsed_ms,
"attempts": attempts,
}
@app.post("/api/tools/whatsminer-read-test")
def whatsminer_read_test(request: WhatsminerReadTestRequest, current_user: dict = Depends(get_current_user)):
result = send_whatsminer_read_command(
ip_address=request.ip_address,
port=request.port,
command=request.command,
timeout_ms=request.timeout_ms,
params=request.params,
)
return api_response(data=result)
@app.get("/api/miners")
def list_miners(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
miners.id::text,
miners.site_id::text,
sites.name AS site_name,
miners.room_id::text,
rooms.name AS room_name,
miners.asset_tag,
miners.current_mac,
host(miners.current_ip) AS current_ip,
miners.hostname,
miners.miner_sn,
miners.power_sn,
miners.model,
miners.miner_type,
miners.api_version,
miners.firmware_version,
miners.platform,
miners.chip,
miners.status,
miners.first_seen_at,
miners.last_seen_at,
miners.updated_at
FROM miners
LEFT JOIN sites ON sites.id = miners.site_id
LEFT JOIN rooms ON rooms.id = miners.room_id
ORDER BY rooms.sort_order, miners.current_ip, miners.current_mac;
"""
)
miners = [dict(row) for row in cur.fetchall()]
return api_response(data=miners)
finally:
if conn is not None:
conn.close()
@app.get("/api/miners/{miner_id}")
def get_miner(miner_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
miners.id::text,
miners.site_id::text,
sites.name AS site_name,
miners.room_id::text,
rooms.name AS room_name,
miners.asset_tag,
miners.current_mac,
host(miners.current_ip) AS current_ip,
miners.hostname,
miners.miner_sn,
miners.power_sn,
miners.model,
miners.miner_type,
miners.api_version,
miners.firmware_version,
miners.platform,
miners.chip,
miners.status,
miners.notes,
miners.first_seen_at,
miners.last_seen_at,
miners.created_at,
miners.updated_at
FROM miners
LEFT JOIN sites ON sites.id = miners.site_id
LEFT JOIN rooms ON rooms.id = miners.room_id
WHERE miners.id = %s;
""",
(miner_id,),
)
miner = cur.fetchone()
if not miner:
raise HTTPException(status_code=404, detail="Miner not found")
return api_response(data=dict(miner))
finally:
if conn is not None:
conn.close()
@app.get("/api/miners/{miner_id}/network-identities")
def list_miner_network_identities(miner_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
miner_id::text,
mac_address,
host(ip_address) AS ip_address,
hostname,
source,
is_current,
first_seen_at,
last_seen_at,
created_at,
updated_at
FROM miner_network_identities
WHERE miner_id = %s
ORDER BY is_current DESC, last_seen_at DESC;
""",
(miner_id,),
)
identities = [dict(row) for row in cur.fetchall()]
return api_response(data=identities)
finally:
if conn is not None:
conn.close()
@app.get("/api/miners/{miner_id}/hashboards")
def list_miner_hashboards(miner_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
miner_id::text,
slot,
pcb_sn,
chip_data,
temperature_c,
chip_frequency,
mhs_av,
mhs_1m,
mhs_15m,
hs_rt,
factory_ghs,
effective_chips,
upfreq_complete,
first_seen_at,
last_seen_at,
updated_at
FROM miner_hashboards
WHERE miner_id = %s
ORDER BY slot;
""",
(miner_id,),
)
hashboards = [dict(row) for row in cur.fetchall()]
return api_response(data=hashboards)
finally:
if conn is not None:
conn.close()
@app.get("/api/miners/{miner_id}/status-snapshots")
def list_miner_status_snapshots(miner_id: str, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
miner_id::text,
host(observed_ip) AS observed_ip,
observed_at,
mhs_av,
mhs_1m,
mhs_15m,
hs_rt,
power_w,
power_rate_w_per_th,
env_temp_c,
chip_temp_min_c,
chip_temp_max_c,
chip_temp_avg_c,
fan_speed_in_rpm,
fan_speed_out_rpm,
uptime_seconds,
hash_stable,
pool_rejected_percent,
created_at
FROM miner_status_snapshots
WHERE miner_id = %s
ORDER BY observed_at DESC
LIMIT 100;
""",
(miner_id,),
)
snapshots = [dict(row) for row in cur.fetchall()]
return api_response(data=snapshots)
finally:
if conn is not None:
conn.close()
class DiscoverOneMinerRequest(BaseModel):
ip_address: str
port: int = Field(default=4028, ge=1, le=65535)
scan_range_id: str | None = None
timeout_ms: int = Field(default=5000, ge=500, le=30000)
def clean_ip_text(value: Any) -> str | None:
if value is None:
return None
return str(value).split("/")[0]
def to_bool_or_none(value: Any) -> bool | None:
if value is None:
return None
if isinstance(value, bool):
return value
text = str(value).strip().lower()
if text in ("true", "1", "yes", "y", "on"):
return True
if text in ("false", "0", "no", "n", "off"):
return False
return None
def get_msg_dict(result: dict, command_name: str) -> dict:
if not result.get("success"):
raise HTTPException(status_code=502, detail=f"{command_name} failed: {result}")
parsed_json = result.get("parsed_json")
if not isinstance(parsed_json, dict):
raise HTTPException(status_code=502, detail=f"{command_name} did not return JSON")
if parsed_json.get("STATUS") == "E":
raise HTTPException(status_code=502, detail=f"{command_name} returned error: {parsed_json}")
msg = parsed_json.get("Msg", {})
if isinstance(msg, dict):
return msg
return {}
def find_scan_range_context(cur, ip_address: str, scan_range_id: str | None) -> dict:
if scan_range_id:
cur.execute(
"""
SELECT
scan_ranges.id::text AS scan_range_id,
scan_ranges.room_id::text,
rooms.site_id::text,
rooms.name AS room_name,
scan_ranges.name AS scan_range_name
FROM scan_ranges
LEFT JOIN rooms ON rooms.id = scan_ranges.room_id
WHERE scan_ranges.id = %s;
""",
(scan_range_id,),
)
row = cur.fetchone()
if row:
return dict(row)
cur.execute(
"""
SELECT
scan_ranges.id::text AS scan_range_id,
scan_ranges.room_id::text,
rooms.site_id::text,
rooms.name AS room_name,
scan_ranges.name AS scan_range_name,
scan_ranges.range_type,
scan_ranges.cidr,
host(scan_ranges.start_ip) AS start_ip,
scan_ranges.end_ip::text
FROM scan_ranges
LEFT JOIN rooms ON rooms.id = scan_ranges.room_id
WHERE scan_ranges.enabled = true;
"""
)
ip_obj = ipaddress.ip_address(ip_address)
for row in cur.fetchall():
row_dict = dict(row)
if row_dict["range_type"] == "cidr" and row_dict["cidr"]:
network = ipaddress.ip_network(row_dict["cidr"], strict=False)
if ip_obj in network:
return row_dict
if row_dict["range_type"] == "start_end" and row_dict["start_ip"] and row_dict["end_ip"]:
start_ip = ipaddress.ip_address(clean_ip_text(row_dict["start_ip"]))
end_ip = ipaddress.ip_address(clean_ip_text(row_dict["end_ip"]))
if int(start_ip) <= int(ip_obj) <= int(end_ip):
return row_dict
return {
"scan_range_id": scan_range_id,
"room_id": None,
"site_id": None,
"room_name": None,
"scan_range_name": None,
}
def find_existing_miner(cur, miner_sn: str | None, mac_address: str | None, ip_address: str) -> dict | None:
if miner_sn:
cur.execute(
"""
SELECT id::text
FROM miners
WHERE miner_sn = %s
LIMIT 1;
""",
(miner_sn,),
)
row = cur.fetchone()
if row:
return dict(row)
if mac_address:
cur.execute(
"""
SELECT id::text
FROM miners
WHERE lower(current_mac) = lower(%s)
LIMIT 1;
""",
(mac_address,),
)
row = cur.fetchone()
if row:
return dict(row)
cur.execute(
"""
SELECT id::text
FROM miners
WHERE current_ip = %s
LIMIT 1;
""",
(ip_address,),
)
row = cur.fetchone()
if row:
return dict(row)
return None
@app.post("/api/miners/discover-one")
def discover_one_miner(request: DiscoverOneMinerRequest, current_user: dict = Depends(get_current_user)):
try:
ip_address = str(ipaddress.ip_address(request.ip_address))
except ValueError as exc:
raise HTTPException(status_code=422, detail=f"Invalid IP address: {exc}")
miner_info_result = send_whatsminer_read_command(
ip_address=ip_address,
port=request.port,
command="get_miner_info",
timeout_ms=request.timeout_ms,
params={
"info": "ip,proto,netmask,gateway,dns,hostname,mac,ledstat,minersn,powersn"
},
)
version_result = send_whatsminer_read_command(
ip_address=ip_address,
port=request.port,
command="get_version",
timeout_ms=request.timeout_ms,
)
summary_result = send_whatsminer_read_command(
ip_address=ip_address,
port=request.port,
command="summary",
timeout_ms=request.timeout_ms,
)
edevs_result = send_whatsminer_read_command(
ip_address=ip_address,
port=request.port,
command="edevs",
timeout_ms=request.timeout_ms,
)
pools_result = send_whatsminer_read_command(
ip_address=ip_address,
port=request.port,
command="pools",
timeout_ms=request.timeout_ms,
)
miner_info_json = miner_info_result.get("parsed_json") or {}
version_json = version_result.get("parsed_json") or {}
summary_json = summary_result.get("parsed_json") or {}
edevs_json = edevs_result.get("parsed_json") or {}
pools_json = pools_result.get("parsed_json") or {}
miner_info_msg = get_msg_dict(miner_info_result, "get_miner_info")
version_msg = get_msg_dict(version_result, "get_version")
try:
summary_msg = get_msg_dict(summary_result, "summary")
summary_warning = None
except HTTPException as exc:
summary_msg = {}
summary_warning = exc.detail
mac_address = miner_info_msg.get("mac")
if mac_address:
mac_address = str(mac_address).upper()
hostname = miner_info_msg.get("hostname")
miner_sn = miner_info_msg.get("minersn")
power_sn = miner_info_msg.get("powersn")
miner_type = version_msg.get("miner_type")
model = miner_type.split("_")[0] if miner_type else None
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
context = find_scan_range_context(cur, ip_address, request.scan_range_id)
existing_miner = find_existing_miner(
cur=cur,
miner_sn=miner_sn,
mac_address=mac_address,
ip_address=ip_address,
)
if existing_miner:
miner_id = existing_miner["id"]
cur.execute(
"""
UPDATE miners
SET site_id = COALESCE(%s, site_id),
room_id = COALESCE(%s, room_id),
current_mac = %s,
current_ip = %s,
hostname = %s,
miner_sn = %s,
power_sn = %s,
model = %s,
miner_type = %s,
api_version = %s,
firmware_version = %s,
platform = %s,
chip = %s,
status = 'online',
last_seen_at = now(),
updated_at = now()
WHERE id = %s
RETURNING id::text;
""",
(
context.get("site_id"),
context.get("room_id"),
mac_address,
ip_address,
hostname,
miner_sn,
power_sn,
model,
miner_type,
version_msg.get("api_ver"),
version_msg.get("fw_ver"),
version_msg.get("platform"),
version_msg.get("chip"),
miner_id,
),
)
miner_id = cur.fetchone()["id"]
action = "miner.updated"
else:
cur.execute(
"""
INSERT INTO miners (
site_id,
room_id,
current_mac,
current_ip,
hostname,
miner_sn,
power_sn,
model,
miner_type,
api_version,
firmware_version,
platform,
chip,
status
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'online')
RETURNING id::text;
""",
(
context.get("site_id"),
context.get("room_id"),
mac_address,
ip_address,
hostname,
miner_sn,
power_sn,
model,
miner_type,
version_msg.get("api_ver"),
version_msg.get("fw_ver"),
version_msg.get("platform"),
version_msg.get("chip"),
),
)
miner_id = cur.fetchone()["id"]
action = "miner.created"
if mac_address:
cur.execute(
"""
UPDATE miner_network_identities
SET is_current = false,
updated_at = now()
WHERE miner_id = %s;
""",
(miner_id,),
)
cur.execute(
"""
SELECT id::text
FROM miner_network_identities
WHERE miner_id = %s
AND lower(mac_address) = lower(%s)
LIMIT 1;
""",
(miner_id, mac_address),
)
identity = cur.fetchone()
if identity:
cur.execute(
"""
UPDATE miner_network_identities
SET ip_address = %s,
hostname = %s,
source = 'scanner',
is_current = true,
last_seen_at = now(),
updated_at = now()
WHERE id = %s;
""",
(ip_address, hostname, identity["id"]),
)
else:
cur.execute(
"""
INSERT INTO miner_network_identities (
miner_id,
mac_address,
ip_address,
hostname,
source,
is_current
)
VALUES (%s, %s, %s, %s, 'scanner', true);
""",
(miner_id, mac_address, ip_address, hostname),
)
for board in edevs_json.get("DEVS", []):
slot = board.get("Slot")
if slot is None:
continue
cur.execute(
"""
INSERT INTO miner_hashboards (
miner_id,
slot,
pcb_sn,
chip_data,
temperature_c,
chip_frequency,
mhs_av,
mhs_1m,
mhs_15m,
hs_rt,
factory_ghs,
effective_chips,
upfreq_complete,
raw_json
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT ON CONSTRAINT uq_miner_hashboards_miner_slot
DO UPDATE SET
pcb_sn = EXCLUDED.pcb_sn,
chip_data = EXCLUDED.chip_data,
temperature_c = EXCLUDED.temperature_c,
chip_frequency = EXCLUDED.chip_frequency,
mhs_av = EXCLUDED.mhs_av,
mhs_1m = EXCLUDED.mhs_1m,
mhs_15m = EXCLUDED.mhs_15m,
hs_rt = EXCLUDED.hs_rt,
factory_ghs = EXCLUDED.factory_ghs,
effective_chips = EXCLUDED.effective_chips,
upfreq_complete = EXCLUDED.upfreq_complete,
raw_json = EXCLUDED.raw_json,
last_seen_at = now(),
updated_at = now();
""",
(
miner_id,
int(slot),
board.get("PCB SN"),
board.get("Chip Data"),
board.get("Temperature"),
board.get("Chip Frequency"),
board.get("MHS av"),
board.get("MHS 1m"),
board.get("MHS 15m"),
board.get("HS RT"),
board.get("Factory GHS"),
board.get("Effective Chips"),
board.get("Upfreq Complete"),
psycopg.types.json.Jsonb(json_safe(board)),
),
)
cur.execute(
"""
INSERT INTO miner_status_snapshots (
miner_id,
observed_ip,
mhs_av,
mhs_1m,
mhs_15m,
hs_rt,
power_w,
power_rate_w_per_th,
env_temp_c,
chip_temp_min_c,
chip_temp_max_c,
chip_temp_avg_c,
fan_speed_in_rpm,
fan_speed_out_rpm,
uptime_seconds,
hash_stable,
pool_rejected_percent,
raw_summary_json,
raw_version_json,
raw_miner_info_json
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);
""",
(
miner_id,
ip_address,
summary_msg.get("MHS av"),
summary_msg.get("MHS 1m"),
summary_msg.get("MHS 15m"),
summary_msg.get("HS RT"),
summary_msg.get("Power"),
summary_msg.get("Power Rate"),
summary_msg.get("Env Temp"),
summary_msg.get("Chip Temp Min"),
summary_msg.get("Chip Temp Max"),
summary_msg.get("Chip Temp Avg"),
summary_msg.get("Fan Speed In"),
summary_msg.get("Fan Speed Out"),
summary_msg.get("Uptime"),
to_bool_or_none(summary_msg.get("Hash Stable")),
summary_msg.get("Pool Rejected%"),
psycopg.types.json.Jsonb(json_safe(summary_json)),
psycopg.types.json.Jsonb(json_safe(version_json)),
psycopg.types.json.Jsonb(json_safe(miner_info_json)),
),
)
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, %s, 'miner', %s, %s);
""",
(
current_user["id"],
action,
miner_id,
psycopg.types.json.Jsonb(
{
"ip_address": ip_address,
"mac_address": mac_address,
"miner_sn": miner_sn,
"power_sn": power_sn,
"miner_type": miner_type,
"firmware_version": version_msg.get("fw_ver"),
"room_id": context.get("room_id"),
"scan_range_id": context.get("scan_range_id"),
}
),
),
)
cur.execute(
"""
SELECT
miners.id::text,
miners.site_id::text,
sites.name AS site_name,
miners.room_id::text,
rooms.name AS room_name,
miners.current_mac,
host(miners.current_ip) AS current_ip,
miners.hostname,
miners.miner_sn,
miners.power_sn,
miners.model,
miners.miner_type,
miners.api_version,
miners.firmware_version,
miners.platform,
miners.chip,
miners.status,
miners.first_seen_at,
miners.last_seen_at,
miners.updated_at
FROM miners
LEFT JOIN sites ON sites.id = miners.site_id
LEFT JOIN rooms ON rooms.id = miners.room_id
WHERE miners.id = %s;
""",
(miner_id,),
)
miner = dict(cur.fetchone())
conn.commit()
return api_response(
data={
"action": action,
"miner": miner,
"summary": {
"ip_address": ip_address,
"mac_address": mac_address,
"miner_sn": miner_sn,
"power_sn": power_sn,
"miner_type": miner_type,
"firmware_version": version_msg.get("fw_ver"),
"hashboards_seen": len(edevs_json.get("DEVS", [])),
"pools_seen": len(pools_json.get("POOLS", [])),
"hashrate_mhs_av": summary_msg.get("MHS av"),
"power_w": summary_msg.get("Power"),
"chip_temp_max_c": summary_msg.get("Chip Temp Max"),
"summary_warning": summary_warning,
},
}
)
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
@app.post("/api/scan-ranges/{scan_range_id}/discover-now")
def discover_scan_range_now(scan_range_id: str, request: ScanNowRequest, current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
id::text,
room_id::text,
name,
range_type,
cidr,
host(start_ip) AS start_ip,
host(end_ip) AS end_ip,
enabled,
timeout_ms,
retry_limit,
parallelism_limit,
stagger_ms,
api_port,
scan_interval_seconds
FROM scan_ranges
WHERE id = %s;
""",
(scan_range_id,),
)
scan_range = cur.fetchone()
if not scan_range:
raise HTTPException(status_code=404, detail="Scan range not found")
if not scan_range["enabled"]:
raise HTTPException(status_code=409, detail="Scan range is disabled")
ip_list = expand_scan_range_for_scan(dict(scan_range), request.max_ips, request.specific_ips)
cur.execute(
"""
INSERT INTO scan_runs (
scan_range_id,
triggered_by,
triggered_by_user_id,
status,
total_ips,
scanned_ips,
miners_found,
errors_count,
notes
)
VALUES (%s, 'manual', %s, 'running', %s, 0, 0, 0, %s)
RETURNING id::text;
""",
(
scan_range_id,
current_user["id"],
len(ip_list),
"TCP port 4028 scan with read-only miner discovery",
),
)
scan_run_id = cur.fetchone()["id"]
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'scan.discovery.started', 'scan_run', %s, %s);
""",
(
current_user["id"],
scan_run_id,
psycopg.types.json.Jsonb(
{
"scan_range_id": scan_range_id,
"scan_range_name": scan_range["name"],
"total_ips": len(ip_list),
"mode": "tcp_4028_then_whatsminer_read_only_discovery",
}
),
),
)
conn.commit()
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
results = []
detected_count = 0
max_workers = min(scan_range["parallelism_limit"], len(ip_list))
if max_workers < 1:
max_workers = 1
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_map = {
executor.submit(
tcp_check_ip,
ip,
scan_range["api_port"],
scan_range["timeout_ms"],
): ip
for ip in ip_list
}
for future in as_completed(future_map):
result = future.result()
results.append(result)
if result["detected"]:
detected_count += 1
results = sorted(results, key=lambda item: ipaddress.ip_address(item["ip_address"]))
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
for result in results:
cur.execute(
"""
INSERT INTO scan_results (
scan_run_id,
ip_address,
port,
detected,
response_ms,
status,
error_message
)
VALUES (%s, %s, %s, %s, %s, %s, %s);
""",
(
scan_run_id,
result["ip_address"],
result["port"],
result["detected"],
result["response_ms"],
result["status"],
result["error_message"],
),
)
cur.execute(
"""
UPDATE scan_ranges
SET last_scan_at = now(),
next_scan_at = now() + make_interval(secs => scan_interval_seconds),
updated_at = now()
WHERE id = %s;
""",
(scan_range_id,),
)
cur.execute(
"""
UPDATE scan_runs
SET scanned_ips = %s,
miners_found = %s,
errors_count = 0
WHERE id = %s;
""",
(
len(results),
detected_count,
scan_run_id,
),
)
conn.commit()
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
discoveries = []
discovery_errors = []
for result in results:
if not result["detected"]:
continue
try:
discovery_response = discover_one_miner(
DiscoverOneMinerRequest(
ip_address=result["ip_address"],
port=result["port"],
scan_range_id=scan_range_id,
timeout_ms=scan_range["timeout_ms"],
),
current_user=current_user,
)
discoveries.append(discovery_response["data"])
except HTTPException as exc:
discovery_errors.append(
{
"ip_address": result["ip_address"],
"status_code": exc.status_code,
"detail": exc.detail,
}
)
except Exception as exc:
discovery_errors.append(
{
"ip_address": result["ip_address"],
"status_code": 500,
"detail": str(exc),
}
)
final_status = "completed"
if discovery_errors:
final_status = "completed_with_errors"
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
UPDATE scan_runs
SET finished_at = now(),
status = %s,
miners_found = %s,
errors_count = %s,
notes = %s
WHERE id = %s
RETURNING
id::text,
scan_range_id::text,
triggered_by,
started_at,
finished_at,
status,
total_ips,
scanned_ips,
miners_found,
errors_count,
notes;
""",
(
final_status,
len(discoveries),
len(discovery_errors),
"TCP port 4028 scan with read-only miner discovery",
scan_run_id,
),
)
scan_run = cur.fetchone()
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'scan.discovery.completed', 'scan_run', %s, %s);
""",
(
current_user["id"],
scan_run_id,
psycopg.types.json.Jsonb(
{
"scan_range_id": scan_range_id,
"total_ips": len(ip_list),
"scanned_ips": len(results),
"port_4028_open": detected_count,
"miners_discovered": len(discoveries),
"discovery_errors": len(discovery_errors),
}
),
),
)
conn.commit()
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
return api_response(
data={
"scan_run": dict(scan_run),
"summary": {
"scan_range_id": scan_range_id,
"scan_range_name": scan_range["name"],
"total_ips": len(ip_list),
"scanned_ips": len(results),
"port_4028_open": detected_count,
"miners_discovered": len(discoveries),
"discovery_errors": len(discovery_errors),
},
"discoveries": discoveries,
"discovery_errors": discovery_errors,
}
)
@app.get("/api/dashboard/summary")
def dashboard_summary(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
COUNT(*) AS total_miners,
COUNT(*) FILTER (WHERE status = 'online') AS online_miners,
COUNT(*) FILTER (WHERE current_mac IS NULL OR current_mac = '') AS missing_mac_count,
COUNT(*) FILTER (WHERE miner_type IS NULL OR miner_type = '') AS missing_miner_type_count,
COUNT(*) FILTER (WHERE firmware_version IS NULL OR firmware_version = '') AS missing_firmware_count,
COUNT(*) FILTER (WHERE last_seen_at >= now() - interval '30 minutes') AS seen_last_30_minutes,
COUNT(*) FILTER (WHERE last_seen_at < now() - interval '30 minutes') AS not_seen_last_30_minutes
FROM miners;
"""
)
totals = dict(cur.fetchone())
cur.execute(
"""
SELECT
COALESCE(rooms.name, 'Unassigned') AS room_name,
COUNT(miners.id) AS miner_count,
COUNT(miners.id) FILTER (WHERE miners.status = 'online') AS online_count,
COUNT(miners.id) FILTER (WHERE miners.last_seen_at >= now() - interval '30 minutes') AS seen_last_30_minutes
FROM miners
LEFT JOIN rooms ON rooms.id = miners.room_id
GROUP BY rooms.name, rooms.sort_order
ORDER BY rooms.sort_order, rooms.name;
"""
)
by_room = [dict(row) for row in cur.fetchall()]
cur.execute(
"""
SELECT
COALESCE(NULLIF(miner_type, ''), NULLIF(model, ''), 'Unknown') AS model,
COUNT(*) AS miner_count
FROM miners
GROUP BY COALESCE(NULLIF(miner_type, ''), NULLIF(model, ''), 'Unknown')
ORDER BY miner_count DESC, model;
"""
)
by_model = [dict(row) for row in cur.fetchall()]
cur.execute(
"""
SELECT
COALESCE(NULLIF(firmware_version, ''), 'Unknown') AS firmware_version,
COUNT(*) AS miner_count
FROM miners
GROUP BY COALESCE(NULLIF(firmware_version, ''), 'Unknown')
ORDER BY miner_count DESC, firmware_version;
"""
)
by_firmware = [dict(row) for row in cur.fetchall()]
cur.execute(
"""
SELECT
COALESCE(NULLIF(status, ''), 'Unknown') AS status,
COUNT(*) AS miner_count
FROM miners
GROUP BY COALESCE(NULLIF(status, ''), 'Unknown')
ORDER BY miner_count DESC, status;
"""
)
by_status = [dict(row) for row in cur.fetchall()]
cur.execute(
"""
WITH latest AS (
SELECT DISTINCT ON (miner_id)
miner_id,
observed_at,
power_w,
power_rate_w_per_th,
chip_temp_max_c,
chip_temp_avg_c,
mhs_av,
hash_stable
FROM miner_status_snapshots
ORDER BY miner_id, observed_at DESC
)
SELECT
COUNT(*) AS miners_with_snapshots,
ROUND(AVG(power_w)::numeric, 2)::float AS avg_power_w,
ROUND(SUM(power_w)::numeric, 2)::float AS total_power_w,
ROUND(AVG(power_rate_w_per_th)::numeric, 2)::float AS avg_power_rate_w_per_th,
ROUND(MAX(chip_temp_max_c)::numeric, 2)::float AS hottest_chip_temp_c,
ROUND(AVG(chip_temp_avg_c)::numeric, 2)::float AS avg_chip_temp_c,
ROUND(SUM(mhs_av)::numeric / 1000000, 2)::float AS total_hashrate_ths,
COUNT(*) FILTER (WHERE chip_temp_max_c >= 90) AS miners_chip_temp_90c_or_higher,
COUNT(*) FILTER (WHERE hash_stable = false) AS miners_hash_unstable
FROM latest;
"""
)
telemetry = dict(cur.fetchone())
cur.execute(
"""
SELECT
COUNT(DISTINCT miner_id) AS miners_with_hashboards,
COUNT(*) AS total_hashboards,
COUNT(*) FILTER (WHERE temperature_c >= 90) AS hashboards_90c_or_higher,
COUNT(*) FILTER (WHERE pcb_sn IS NULL OR pcb_sn = '') AS hashboards_missing_pcb_sn
FROM miner_hashboards;
"""
)
hashboards = dict(cur.fetchone())
return api_response(
data={
"totals": totals,
"by_room": by_room,
"by_model": by_model,
"by_firmware": by_firmware,
"by_status": by_status,
"telemetry": telemetry,
"hashboards": hashboards,
}
)
finally:
if conn is not None:
conn.close()
from fastapi import File, Query, UploadFile
import csv
import io
def csv_value(row: dict, possible_names: list[str]) -> str | None:
for name in possible_names:
if name in row and row[name] is not None:
value = str(row[name]).strip()
if value:
return value
return None
def normalize_mac_for_match(mac: str | None) -> str | None:
if not mac:
return None
cleaned = "".join(ch for ch in mac.lower() if ch in "0123456789abcdef")
if len(cleaned) != 12:
return None
return cleaned
def model_from_miner_type(miner_type: str | None) -> str | None:
if not miner_type:
return None
if "_" in miner_type:
return miner_type.split("_", 1)[0]
return miner_type
@app.post("/api/imports/whatsminer-tool-scan")
async def import_whatsminer_tool_scan(
dry_run: bool = Query(default=True),
update_existing: bool = Query(default=False),
file: UploadFile = File(...),
current_user: dict = Depends(get_current_user),
):
file_bytes = await file.read()
try:
csv_text = file_bytes.decode("utf-8-sig")
except UnicodeDecodeError:
csv_text = file_bytes.decode("latin-1")
reader = csv.DictReader(io.StringIO(csv_text))
rows = list(reader)
if not rows:
raise HTTPException(status_code=422, detail="CSV file has no rows")
matched_count = 0
unmatched_count = 0
rows_with_updates = 0
updates_applied = 0
skipped_no_useful_data = 0
update_samples = []
unmatched_samples = []
conn = None
try:
conn = get_db_connection()
conn.autocommit = False
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
for row_number, row in enumerate(rows, start=2):
ip_address = csv_value(row, ["IP", "IP Address", "ip", "ip_address"])
mac_address = csv_value(row, ["MAC Addr", "MAC", "Mac", "mac", "mac_address"])
miner_type = csv_value(row, ["Miner Type", "MinerType", "Model", "model", "miner_type"])
firmware_version = csv_value(row, ["Firmware", "Firmware Version", "FW Version", "fw_ver", "firmware_version"])
if ip_address:
ip_address = clean_ip_text(ip_address)
normalized_mac = normalize_mac_for_match(mac_address)
model = model_from_miner_type(miner_type)
if not ip_address and not normalized_mac:
skipped_no_useful_data += 1
continue
miner = None
if normalized_mac:
cur.execute(
"""
SELECT
id::text,
host(current_ip) AS current_ip,
current_mac,
model,
miner_type,
firmware_version
FROM miners
WHERE regexp_replace(lower(current_mac), '[^0-9a-f]', '', 'g') = %s
LIMIT 1;
""",
(normalized_mac,),
)
miner = cur.fetchone()
if not miner and ip_address:
cur.execute(
"""
SELECT
id::text,
host(current_ip) AS current_ip,
current_mac,
model,
miner_type,
firmware_version
FROM miners
WHERE host(current_ip) = %s
LIMIT 1;
""",
(ip_address,),
)
miner = cur.fetchone()
if not miner:
unmatched_count += 1
if len(unmatched_samples) < 20:
unmatched_samples.append(
{
"row_number": row_number,
"ip_address": ip_address,
"mac_address": mac_address,
"miner_type": miner_type,
"firmware_version": firmware_version,
}
)
continue
matched_count += 1
set_clauses = []
params = []
proposed_changes = {}
if miner_type and (update_existing or not miner["miner_type"]):
set_clauses.append("miner_type = %s")
params.append(miner_type)
proposed_changes["miner_type"] = {
"old": miner["miner_type"],
"new": miner_type,
}
if model and (update_existing or not miner["model"]):
set_clauses.append("model = %s")
params.append(model)
proposed_changes["model"] = {
"old": miner["model"],
"new": model,
}
if firmware_version and (update_existing or not miner["firmware_version"]):
set_clauses.append("firmware_version = %s")
params.append(firmware_version)
proposed_changes["firmware_version"] = {
"old": miner["firmware_version"],
"new": firmware_version,
}
if not set_clauses:
continue
rows_with_updates += 1
if len(update_samples) < 50:
update_samples.append(
{
"row_number": row_number,
"miner_id": miner["id"],
"ip_address": miner["current_ip"],
"current_mac": miner["current_mac"],
"changes": proposed_changes,
}
)
if not dry_run:
set_sql = ", ".join(set_clauses)
params.append(miner["id"])
cur.execute(
f"""
UPDATE miners
SET {set_sql},
updated_at = now()
WHERE id = %s;
""",
tuple(params),
)
updates_applied += 1
if not dry_run:
cur.execute(
"""
INSERT INTO audit_logs (actor_user_id, action, entity_type, entity_id, new_value_json)
VALUES (%s, 'import.whatsminertool_csv.applied', 'import', NULL, %s);
""",
(
current_user["id"],
psycopg.types.json.Jsonb(
{
"filename": file.filename,
"rows_read": len(rows),
"matched_count": matched_count,
"unmatched_count": unmatched_count,
"rows_with_updates": rows_with_updates,
"updates_applied": updates_applied,
"update_existing": update_existing,
}
),
),
)
if dry_run:
conn.rollback()
else:
conn.commit()
return api_response(
data={
"filename": file.filename,
"dry_run": dry_run,
"update_existing": update_existing,
"rows_read": len(rows),
"matched_count": matched_count,
"unmatched_count": unmatched_count,
"skipped_no_useful_data": skipped_no_useful_data,
"rows_with_updates": rows_with_updates,
"updates_applied": updates_applied,
"update_samples": update_samples,
"unmatched_samples": unmatched_samples,
}
)
except HTTPException:
if conn is not None:
conn.rollback()
raise
except Exception as exc:
if conn is not None:
conn.rollback()
raise HTTPException(status_code=500, detail=str(exc))
finally:
if conn is not None:
conn.close()
from fastapi.responses import HTMLResponse
@app.get("/tools/whatsminer-import", response_class=HTMLResponse)
def whatsminer_import_tool_page():
return """
MinerOps - WhatsMinerTool CSV Import
New miner WebAPI reminder: Before importing or adding new WhatsMiners, enable WebAPI/API3 in WhatsMinerTool. Reboot, Suspend, Resume, and password changes need WebAPI enabled.
MinerOps - WhatsMinerTool CSV Import
1. Login
Not logged in.
2. Upload CSV
Choose a CSV and preview first.
3. Preview Summary
Rows Read
-
Matched
-
Unmatched
-
Rows With Updates
-
Updates Applied
-
Raw Result
No result yet.
"""
@app.get("/api/dashboard/miner-table")
def dashboard_miner_table(current_user: dict = Depends(get_current_user)):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
WITH latest AS (
SELECT DISTINCT ON (miner_id)
miner_id,
observed_at,
observed_ip,
mhs_av,
power_w,
power_rate_w_per_th,
chip_temp_max_c,
chip_temp_avg_c,
fan_speed_in_rpm,
fan_speed_out_rpm,
hash_stable,
pool_rejected_percent
FROM miner_status_snapshots
ORDER BY miner_id, observed_at DESC
)
SELECT
miners.id::text,
COALESCE(rooms.name, 'Unassigned') AS room_name,
host(miners.current_ip) AS current_ip,
miners.current_mac,
miners.hostname,
miners.miner_sn,
miners.power_sn,
miners.model,
miners.miner_type,
miners.firmware_version,
miners.status,
miners.last_seen_at,
latest.observed_at,
ROUND((latest.mhs_av / 1000000.0)::numeric, 2)::float AS hashrate_ths,
ROUND(latest.power_w::numeric, 2)::float AS power_w,
ROUND(latest.power_rate_w_per_th::numeric, 2)::float AS power_rate_w_per_th,
ROUND(latest.chip_temp_max_c::numeric, 2)::float AS chip_temp_max_c,
ROUND(latest.chip_temp_avg_c::numeric, 2)::float AS chip_temp_avg_c,
latest.fan_speed_in_rpm,
latest.fan_speed_out_rpm,
latest.hash_stable,
ROUND(latest.pool_rejected_percent::numeric, 4)::float AS pool_rejected_percent
FROM miners
LEFT JOIN rooms ON rooms.id = miners.room_id
LEFT JOIN latest ON latest.miner_id = miners.id
ORDER BY rooms.sort_order, miners.current_ip;
"""
)
rows = [dict(row) for row in cur.fetchall()]
return api_response(data=rows)
finally:
if conn is not None:
conn.close()
@app.get("/api/dashboard/hot-miners")
def dashboard_hot_miners(
min_temp_c: float = 90,
limit: int = 100,
current_user: dict = Depends(get_current_user),
):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
WITH latest AS (
SELECT DISTINCT ON (miner_id)
miner_id,
observed_at,
chip_temp_max_c,
chip_temp_avg_c,
power_w,
mhs_av,
hash_stable
FROM miner_status_snapshots
ORDER BY miner_id, observed_at DESC
)
SELECT
miners.id::text,
COALESCE(rooms.name, 'Unassigned') AS room_name,
host(miners.current_ip) AS current_ip,
miners.current_mac,
miners.model,
miners.miner_type,
miners.firmware_version,
miners.status,
latest.observed_at,
ROUND(latest.chip_temp_max_c::numeric, 2)::float AS chip_temp_max_c,
ROUND(latest.chip_temp_avg_c::numeric, 2)::float AS chip_temp_avg_c,
ROUND(latest.power_w::numeric, 2)::float AS power_w,
ROUND((latest.mhs_av / 1000000.0)::numeric, 2)::float AS hashrate_ths,
latest.hash_stable
FROM miners
LEFT JOIN rooms ON rooms.id = miners.room_id
JOIN latest ON latest.miner_id = miners.id
WHERE latest.chip_temp_max_c >= %s
ORDER BY latest.chip_temp_max_c DESC
LIMIT %s;
""",
(min_temp_c, limit),
)
rows = [dict(row) for row in cur.fetchall()]
return api_response(data=rows)
finally:
if conn is not None:
conn.close()
@app.get("/api/dashboard/unstable-miners")
def dashboard_unstable_miners(
limit: int = 100,
current_user: dict = Depends(get_current_user),
):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
WITH latest AS (
SELECT DISTINCT ON (miner_id)
miner_id,
observed_at,
chip_temp_max_c,
power_w,
mhs_av,
hash_stable,
pool_rejected_percent
FROM miner_status_snapshots
ORDER BY miner_id, observed_at DESC
)
SELECT
miners.id::text,
COALESCE(rooms.name, 'Unassigned') AS room_name,
host(miners.current_ip) AS current_ip,
miners.current_mac,
miners.model,
miners.miner_type,
miners.firmware_version,
miners.status,
latest.observed_at,
ROUND((latest.mhs_av / 1000000.0)::numeric, 2)::float AS hashrate_ths,
ROUND(latest.power_w::numeric, 2)::float AS power_w,
ROUND(latest.chip_temp_max_c::numeric, 2)::float AS chip_temp_max_c,
latest.hash_stable,
ROUND(latest.pool_rejected_percent::numeric, 4)::float AS pool_rejected_percent
FROM miners
LEFT JOIN rooms ON rooms.id = miners.room_id
JOIN latest ON latest.miner_id = miners.id
WHERE latest.hash_stable = false
OR latest.pool_rejected_percent >= 1
ORDER BY latest.hash_stable ASC, latest.pool_rejected_percent DESC NULLS LAST, latest.chip_temp_max_c DESC NULLS LAST
LIMIT %s;
""",
(limit,),
)
rows = [dict(row) for row in cur.fetchall()]
return api_response(data=rows)
finally:
if conn is not None:
conn.close()
@app.get("/api/dashboard/stale-miners")
def dashboard_stale_miners(
minutes: int = 30,
limit: int = 200,
current_user: dict = Depends(get_current_user),
):
conn = None
try:
conn = get_db_connection()
with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
cur.execute(
"""
SELECT
miners.id::text,
COALESCE(rooms.name, 'Unassigned') AS room_name,
host(miners.current_ip) AS current_ip,
miners.current_mac,
miners.model,
miners.miner_type,
miners.firmware_version,
miners.status,
miners.last_seen_at,
ROUND(EXTRACT(EPOCH FROM (now() - miners.last_seen_at)) / 60.0, 1)::float AS minutes_since_seen
FROM miners
LEFT JOIN rooms ON rooms.id = miners.room_id
WHERE miners.last_seen_at < now() - make_interval(mins => %s)
ORDER BY miners.last_seen_at ASC
LIMIT %s;
""",
(minutes, limit),
)
rows = [dict(row) for row in cur.fetchall()]
return api_response(data=rows)
finally:
if conn is not None:
conn.close()
MINEROPS_UNIFIED_CSS = r"""
:root {
--minerops-bg: #f4f7fb;
--minerops-card: #ffffff;
--minerops-border: #d9e2ec;
--minerops-text: #102a43;
--minerops-muted: #52606d;
--minerops-blue: #0b63ce;
--minerops-blue-dark: #084b9a;
--minerops-green: #0f766e;
--minerops-red: #dc2626;
}
body {
background: var(--minerops-bg) !important;
color: var(--minerops-text) !important;
font-family: "Segoe UI", Arial, sans-serif !important;
margin: 0 !important;
padding: 20px !important;
}
/* Hide old page headers/navs that made the pages look duplicated */
body > header:not(.minerops-shell-header) {
display: none !important;
}
body > h1 {
display: none !important;
}
body > .topbar {
display: none !important;
}
.app-header {
display: none !important;
}
#dashboardLocationMapLink {
display: none !important;
}
.minerops-shell-header {
background: linear-gradient(135deg, #0b63ce, #0f766e);
color: white;
border-radius: 16px;
padding: 18px 20px;
margin-bottom: 16px;
display: flex;
justify-content: space-between;
align-items: center;
gap: 16px;
box-shadow: 0 8px 18px rgba(15, 23, 42, 0.18);
}
.minerops-kicker {
font-size: 13px;
font-weight: 800;
letter-spacing: 0.08em;
text-transform: uppercase;
opacity: 0.86;
margin-bottom: 4px;
}
.minerops-shell-header h1 {
display: block !important;
margin: 0;
color: white;
font-size: 28px;
line-height: 1.1;
}
.minerops-subtitle {
margin-top: 6px;
font-size: 14px;
opacity: 0.9;
}
.minerops-nav {
display: flex;
flex-wrap: wrap;
justify-content: flex-end;
gap: 8px;
}
.minerops-nav-link {
display: inline-block;
border: 1px solid rgba(255, 255, 255, 0.45);
background: rgba(255, 255, 255, 0.14);
color: white !important;
border-radius: 999px;
padding: 8px 12px;
font-weight: 800;
font-size: 13px;
text-decoration: none !important;
}
.minerops-nav-link:hover {
background: rgba(255, 255, 255, 0.24);
}
.minerops-nav-link.active {
background: white;
color: var(--minerops-blue) !important;
}
.card {
background: var(--minerops-card) !important;
border: 1px solid var(--minerops-border) !important;
border-radius: 14px !important;
padding: 16px !important;
margin-bottom: 16px !important;
box-shadow: 0 2px 8px rgba(15, 23, 42, 0.06) !important;
}
.card h2 {
color: var(--minerops-text) !important;
margin-top: 0 !important;
}
.small,
.muted {
color: var(--minerops-muted) !important;
}
button,
input[type="button"],
input[type="submit"],
.issue-action-button {
background: var(--minerops-blue) !important;
color: white !important;
border: 0 !important;
border-radius: 8px !important;
padding: 8px 11px !important;
font-weight: 800 !important;
cursor: pointer !important;
text-decoration: none !important;
}
button:hover,
input[type="button"]:hover,
input[type="submit"]:hover,
.issue-action-button:hover {
background: var(--minerops-blue-dark) !important;
}
button.secondary {
background: #64748b !important;
}
button.danger,
.danger {
background: var(--minerops-red) !important;
color: white !important;
}
input,
select,
textarea {
border: 1px solid #bcccdc !important;
border-radius: 8px !important;
padding: 8px !important;
background: white !important;
color: var(--minerops-text) !important;
}
table {
width: 100%;
border-collapse: collapse;
background: white;
border-radius: 10px;
overflow: hidden;
}
th {
background: #eef2ff !important;
color: #1e293b !important;
text-align: left;
font-size: 13px;
}
th,
td {
border-bottom: 1px solid #e2e8f0 !important;
padding: 8px !important;
vertical-align: top;
}
tr:hover td {
background: #f8fafc !important;
}
.status {
border-radius: 10px !important;
padding: 10px 12px !important;
margin-top: 10px !important;
border: 1px solid var(--minerops-border) !important;
background: #f8fafc !important;
}
.status.ok {
background: #ecfdf5 !important;
border-color: #6ee7b7 !important;
}
.status.warn {
background: #fff7ed !important;
border-color: #fdba74 !important;
}
.status.bad {
background: #fef2f2 !important;
border-color: #fca5a5 !important;
}
.sticky-mini-toolbar,
.card-toolbar {
background: rgba(255, 255, 255, 0.96) !important;
border: 1px solid var(--minerops-border) !important;
border-radius: 14px !important;
box-shadow: 0 4px 12px rgba(15, 23, 42, 0.10) !important;
padding: 10px !important;
}
.sticky-mini-toolbar {
top: 8px !important;
}
/* Step 80: Header auth controls */
.minerops-auth {
display: flex;
flex-wrap: wrap;
align-items: center;
gap: 6px;
margin-left: 8px;
}
.minerops-auth input {
width: 115px !important;
min-width: 90px !important;
padding: 7px 8px !important;
border-radius: 999px !important;
border: 1px solid rgba(255, 255, 255, 0.55) !important;
background: rgba(255, 255, 255, 0.92) !important;
font-size: 12px !important;
}
.minerops-auth button {
border-radius: 999px !important;
padding: 7px 10px !important;
font-size: 12px !important;
background: rgba(255, 255, 255, 0.18) !important;
border: 1px solid rgba(255, 255, 255, 0.45) !important;
color: white !important;
}
.minerops-auth button:hover {
background: rgba(255, 255, 255, 0.28) !important;
}
.minerops-auth-user {
color: white;
font-size: 13px;
font-weight: 800;
padding: 7px 10px;
border-radius: 999px;
background: rgba(255, 255, 255, 0.14);
border: 1px solid rgba(255, 255, 255, 0.30);
}
.minerops-auth-message {
color: white;
font-size: 12px;
font-weight: 700;
opacity: 0.92;
}
.minerops-hidden-login-card {
display: none !important;
}
@media (max-width: 950px) {
.minerops-auth {
width: 100%;
margin-left: 0;
}
.minerops-auth input {
width: 140px !important;
}
}
/* Step 83: Global miner search */
.minerops-global-search {
position: relative;
display: flex;
align-items: center;
gap: 6px;
margin-left: 8px;
}
.minerops-global-search input {
width: 230px !important;
min-width: 180px !important;
border-radius: 999px !important;
padding: 8px 12px !important;
border: 1px solid rgba(255, 255, 255, 0.55) !important;
background: rgba(255, 255, 255, 0.95) !important;
font-size: 13px !important;
}
.minerops-search-results {
position: absolute;
top: 40px;
right: 0;
z-index: 5000;
width: min(680px, 92vw);
max-height: 440px;
overflow: auto;
background: white;
color: var(--minerops-text);
border: 1px solid var(--minerops-border);
border-radius: 14px;
box-shadow: 0 14px 32px rgba(15, 23, 42, 0.28);
padding: 8px;
display: none;
}
.minerops-search-results.open {
display: block;
}
.minerops-search-result {
display: grid;
grid-template-columns: 150px 110px 1fr;
gap: 8px;
padding: 9px;
border-radius: 10px;
cursor: pointer;
border-bottom: 1px solid #e2e8f0;
}
.minerops-search-result:hover,
.minerops-search-result.active {
background: #eff6ff;
}
.minerops-search-result-main {
font-weight: 900;
color: #0f172a;
}
.minerops-search-result-sub {
font-size: 12px;
color: #475569;
}
.minerops-search-empty {
padding: 10px;
color: #475569;
font-size: 13px;
}
@media (max-width: 950px) {
.minerops-global-search {
width: 100%;
margin-left: 0;
}
.minerops-global-search input {
width: 100% !important;
}
.minerops-search-results {
left: 0;
right: auto;
width: 92vw;
}
}
/* Step 84: Global scan action */
.minerops-header-scan {
display: flex;
align-items: center;
gap: 6px;
margin-left: 4px;
}
.minerops-scan-button {
background: rgba(255, 255, 255, 0.18) !important;
border: 1px solid rgba(255, 255, 255, 0.45) !important;
color: white !important;
border-radius: 999px !important;
padding: 8px 12px !important;
font-size: 13px !important;
font-weight: 800 !important;
}
.minerops-scan-button:hover {
background: rgba(255, 255, 255, 0.28) !important;
}
.minerops-scan-button.running {
background: #f59e0b !important;
color: #111827 !important;
}
.minerops-scan-status {
color: white;
font-size: 12px;
font-weight: 700;
opacity: 0.92;
max-width: 260px;
}
@media (max-width: 950px) {
.minerops-header-scan {
width: 100%;
margin-left: 0;
}
.minerops-scan-status {
max-width: 100%;
}
}
@media (max-width: 800px) {
body {
padding: 12px !important;
}
.minerops-shell-header {
align-items: flex-start;
flex-direction: column;
}
.minerops-nav {
justify-content: flex-start;
}
}
"""
@app.get("/assets/minerops-unified.css")
async def minerops_unified_css():
return HTMLResponse(content=MINEROPS_UNIFIED_CSS, media_type="text/css")
MINEROPS_SHELL_AUTH_JS = r"""
(function () {
const TOKEN_KEYS = [
"minerops_token",
"access_token",
"token",
"auth_token",
"authToken"
];
const USERNAME_KEY = "minerops_username";
function escapeHtml(value) {
return String(value || "")
.replaceAll("&", "&")
.replaceAll("<", "<")
.replaceAll(">", ">")
.replaceAll('"', """)
.replaceAll("'", "'");
}
function delay(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}
function looksLikeToken(value) {
if (!value || typeof value !== "string") {
return false;
}
if (value.split(".").length === 3 && value.length > 30) {
return true;
}
return value.length > 40 && !value.includes(" ");
}
function findTokenRecursive(value) {
if (!value) {
return "";
}
if (typeof value === "string") {
return looksLikeToken(value) ? value : "";
}
if (Array.isArray(value)) {
for (const item of value) {
const found = findTokenRecursive(item);
if (found) {
return found;
}
}
return "";
}
if (typeof value === "object") {
const preferredKeys = [
"access_token",
"accessToken",
"token",
"jwt",
"auth_token",
"authToken"
];
for (const key of preferredKeys) {
if (looksLikeToken(value[key])) {
return value[key];
}
}
for (const key of Object.keys(value)) {
if (key.toLowerCase().includes("token")) {
const found = findTokenRecursive(value[key]);
if (found) {
return found;
}
}
}
for (const key of Object.keys(value)) {
const found = findTokenRecursive(value[key]);
if (found) {
return found;
}
}
}
return "";
}
function getToken() {
for (const key of TOKEN_KEYS) {
const value = localStorage.getItem(key);
if (value) {
return value;
}
}
return "";
}
function setToken(token) {
for (const key of TOKEN_KEYS) {
localStorage.setItem(key, token);
}
}
function clearToken() {
for (const key of TOKEN_KEYS) {
localStorage.removeItem(key);
}
localStorage.removeItem(USERNAME_KEY);
}
function authHeaders() {
const token = getToken();
if (!token) {
return {};
}
return {
"Authorization": "Bearer " + token
};
}
function findLegacyLoginContainer(heading) {
let node = heading.parentElement;
let best = null;
for (let i = 0; i < 7 && node && node !== document.body; i++) {
const hasPassword = node.querySelector && node.querySelector('input[type="password"]');
const hasClearToken = node.textContent && node.textContent.toLowerCase().includes("clear token");
const isCard = node.classList && (node.classList.contains("card") || node.classList.contains("panel"));
if (hasPassword || hasClearToken) {
best = node;
}
if (isCard) {
best = node;
break;
}
node = node.parentElement;
}
return best || heading.parentElement;
}
function hideLegacyLoginCards() {
document.querySelectorAll("h1, h2, h3").forEach(heading => {
if (heading.closest(".minerops-shell-header")) {
return;
}
const title = heading.textContent.trim().toLowerCase();
if (
title === "login" ||
title === "login / refresh" ||
title === "login/refresh"
) {
const container = findLegacyLoginContainer(heading);
if (container && !container.closest(".minerops-shell-header")) {
container.classList.add("minerops-hidden-login-card");
container.style.display = "none";
}
}
});
}
function setLegacyLoginInputs(username, password) {
const usernameInputs = Array.from(document.querySelectorAll('input[type="text"], input:not([type])'))
.filter(input => !input.closest(".minerops-shell-header"));
const passwordInputs = Array.from(document.querySelectorAll('input[type="password"]'))
.filter(input => !input.closest(".minerops-shell-header"));
for (const input of usernameInputs) {
const idName = ((input.id || "") + " " + (input.name || "") + " " + (input.placeholder || "")).toLowerCase();
if (idName.includes("user") || idName.includes("admin") || idName.includes("login")) {
input.value = username;
input.dispatchEvent(new Event("input", { bubbles: true }));
input.dispatchEvent(new Event("change", { bubbles: true }));
}
}
for (const input of passwordInputs) {
input.value = password;
input.dispatchEvent(new Event("input", { bubbles: true }));
input.dispatchEvent(new Event("change", { bubbles: true }));
}
}
async function tryLegacyPageLogin(username, password) {
setLegacyLoginInputs(username, password);
const functionNames = [
"login",
"doLogin",
"loginUser",
"dashboardLogin",
"locationLogin"
];
for (const functionName of functionNames) {
if (typeof window[functionName] === "function") {
try {
const result = window[functionName]();
await Promise.resolve(result);
await delay(800);
if (getToken()) {
return true;
}
} catch (error) {
// Keep trying other methods.
}
}
}
const buttons = Array.from(document.querySelectorAll("button"))
.filter(button => !button.closest(".minerops-shell-header"))
.filter(button => button.textContent.trim().toLowerCase() === "login");
for (const button of buttons) {
try {
button.click();
await delay(1000);
if (getToken()) {
return true;
}
} catch (error) {
// Keep trying.
}
}
return false;
}
async function parseLoginResponse(response) {
const text = await response.text();
let data = null;
try {
data = text ? JSON.parse(text) : {};
} catch (error) {
data = text;
}
if (!response.ok) {
const message = data && typeof data === "object"
? (data.detail || data.message || data.error || "Login failed.")
: (text || "Login failed.");
throw new Error(message);
}
return findTokenRecursive(data) || findTokenRecursive(text);
}
async function tryJsonLogin(username, password) {
const response = await fetch("/api/auth/login", {
method: "POST",
headers: {
"Content-Type": "application/json"
},
body: JSON.stringify({
username: username,
password: password
})
});
return await parseLoginResponse(response);
}
async function tryFormLogin(username, password) {
const form = new URLSearchParams();
form.set("username", username);
form.set("password", password);
const response = await fetch("/api/auth/login", {
method: "POST",
headers: {
"Content-Type": "application/x-www-form-urlencoded"
},
body: form.toString()
});
return await parseLoginResponse(response);
}
async function getCurrentUser() {
const token = getToken();
if (!token) {
return null;
}
try {
const response = await fetch("/api/auth/me", {
headers: authHeaders()
});
if (!response.ok) {
return null;
}
return await response.json();
} catch (error) {
return null;
}
}
function userDisplayName(user) {
const stored = localStorage.getItem(USERNAME_KEY);
const candidates = [
stored,
user && user.username,
user && user.user_name,
user && user.name,
user && user.email
];
for (const candidate of candidates) {
if (!candidate) {
continue;
}
const value = String(candidate).trim();
if (!value) {
continue;
}
if (value.toLowerCase() === "signed in") {
continue;
}
return value;
}
return "";
}
async function loginFromHeader() {
const username = document.getElementById("mineropsHeaderUsername")?.value || "";
const password = document.getElementById("mineropsHeaderPassword")?.value || "";
const message = document.getElementById("mineropsHeaderAuthMessage");
if (!username || !password) {
if (message) {
message.textContent = "Enter username and password.";
}
return;
}
if (message) {
message.textContent = "Signing in...";
}
let token = "";
let lastError = null;
try {
token = await tryJsonLogin(username, password);
} catch (error) {
lastError = error;
}
if (!token) {
try {
token = await tryFormLogin(username, password);
} catch (error) {
lastError = error;
}
}
if (token) {
setToken(token);
localStorage.setItem(USERNAME_KEY, username);
window.location.reload();
return;
}
if (message) {
message.textContent = "Trying page login...";
}
const legacyWorked = await tryLegacyPageLogin(username, password);
if (legacyWorked || getToken()) {
localStorage.setItem(USERNAME_KEY, username);
window.location.reload();
return;
}
if (message) {
message.textContent = lastError ? lastError.message : "Login failed. No token was returned.";
}
}
function logoutFromHeader() {
clearToken();
window.location.reload();
}
async function renderHeaderAuth() {
const nav = document.querySelector(".minerops-nav");
if (!nav) {
return;
}
let authArea = document.getElementById("mineropsHeaderAuth");
if (!authArea) {
authArea = document.createElement("div");
authArea.id = "mineropsHeaderAuth";
authArea.className = "minerops-auth";
nav.appendChild(authArea);
}
const user = await getCurrentUser();
if (user) {
const displayName = userDisplayName(user);
authArea.innerHTML = displayName
? `
Signed in: ${escapeHtml(displayName)}
`
: `
Signed in
`;
document.getElementById("mineropsHeaderLogout").onclick = logoutFromHeader;
return;
}
authArea.innerHTML = `
`;
document.getElementById("mineropsHeaderLogin").onclick = loginFromHeader;
document.getElementById("mineropsHeaderPassword").addEventListener("keydown", event => {
if (event.key === "Enter") {
loginFromHeader();
}
});
}
function initHeaderAuth() {
hideLegacyLoginCards();
renderHeaderAuth();
}
if (document.readyState === "loading") {
document.addEventListener("DOMContentLoaded", initHeaderAuth);
} else {
initHeaderAuth();
}
setTimeout(initHeaderAuth, 300);
setTimeout(initHeaderAuth, 1000);
setTimeout(initHeaderAuth, 2000);
window.mineropsHeaderLogout = logoutFromHeader;
window.mineropsHeaderLogin = loginFromHeader;
})();
"""
@app.get("/assets/minerops-shell-auth.js")
async def minerops_shell_auth_js():
return HTMLResponse(content=MINEROPS_SHELL_AUTH_JS, media_type="text/javascript")
MINEROPS_GLOBAL_SEARCH_JS = r"""
(function () {
const TOKEN_KEYS = [
"minerops_token",
"minerops_access_token",
"access_token",
"accessToken",
"token",
"auth_token",
"authToken",
"jwt"
];
let searchCache = null;
let activeIndex = -1;
let currentResults = [];
let lastLoadSummary = "";
function storageGet(key) {
return localStorage.getItem(key) || sessionStorage.getItem(key) || "";
}
function getToken() {
for (const key of TOKEN_KEYS) {
const value = storageGet(key);
if (value) {
return value;
}
}
for (const store of [localStorage, sessionStorage]) {
for (let i = 0; i < store.length; i++) {
const key = store.key(i) || "";
const lower = key.toLowerCase();
if (lower.includes("token") || lower.includes("jwt") || lower.includes("auth")) {
const value = store.getItem(key);
if (value && value.length > 20) {
return value;
}
}
}
}
return "";
}
function authHeaders() {
const token = getToken();
if (!token) {
return {};
}
return {
"Authorization": "Bearer " + token
};
}
function escapeHtml(value) {
return String(value || "")
.replaceAll("&", "&")
.replaceAll("<", "<")
.replaceAll(">", ">")
.replaceAll('"', """)
.replaceAll("'", "'");
}
function normalize(value) {
return String(value || "")
.toLowerCase()
.replaceAll(":", "")
.replaceAll("-", "")
.replaceAll(" ", "")
.trim();
}
function firstValue(obj, keys) {
for (const key of keys) {
if (obj && obj[key] !== null && obj[key] !== undefined && obj[key] !== "") {
return obj[key];
}
}
return "";
}
function valueFromNested(obj, paths) {
for (const path of paths) {
let current = obj;
let found = true;
for (const part of path.split(".")) {
if (!current || current[part] === null || current[part] === undefined || current[part] === "") {
found = false;
break;
}
current = current[part];
}
if (found) {
return current;
}
}
return "";
}
function displayValue(value, fallback) {
if (value === null || value === undefined || value === "") {
return fallback || "";
}
return String(value);
}
async function fetchJson(url) {
const response = await fetch(url, {
headers: authHeaders()
});
if (!response.ok) {
throw new Error(url + " returned HTTP " + response.status);
}
return await response.json();
}
function addSearchRow(rows, item, sourceName) {
if (!item || typeof item !== "object") {
return;
}
const minerId = firstValue(item, [
"miner_id",
"id",
"minerId",
"miner_uuid",
"uuid"
]);
const mac = firstValue(item, [
"current_mac",
"current_mac_address",
"mac_address",
"mac",
"expected_mac",
"network_mac",
"macAddress",
"primary_mac"
]);
const ip = firstValue(item, [
"current_ip",
"current_ip_address",
"ip_address",
"ip",
"network_ip",
"ipAddress",
"host",
"hostname"
]);
const model = firstValue(item, [
"model",
"miner_model",
"miner_type",
"hardware_model",
"machine_model",
"platform"
]);
const room = firstValue(item, [
"room_name",
"room",
"unit",
"unit_name"
]);
const rack = firstValue(item, [
"rack_name",
"rack"
]);
const shelf = firstValue(item, [
"shelf_name",
"shelf"
]);
const position = firstValue(item, [
"position_index",
"position",
"slot",
"slot_index"
]);
const serial = firstValue(item, [
"serial_number",
"miner_serial",
"serial",
"sn",
"control_board_serial",
"control_board_sn",
"psu_serial",
"hashboard_serial"
]) || valueFromNested(item, [
"latest_status.serial_number",
"latest_snapshot.serial_number",
"identity.serial_number"
]);
if (!minerId && !mac && !ip && !serial && !room && !rack) {
return;
}
const locationParts = [];
if (room) {
locationParts.push(room);
}
if (rack) {
locationParts.push(rack);
}
if (shelf) {
locationParts.push(shelf);
}
if (position) {
locationParts.push("Pos " + position);
}
const location = locationParts.join(" / ");
const hashrate = firstValue(item, [
"hashrate_ths",
"hashrate",
"avg_hashrate_ths",
"current_hashrate_ths"
]);
const temp = firstValue(item, [
"chip_temp_max_c",
"max_temp_c",
"temperature_c",
"temp_c"
]);
const haystack = [
minerId,
mac,
ip,
model,
room,
rack,
shelf,
position,
serial,
location,
sourceName
].map(value => String(value || "")).join(" ");
rows.push({
miner_id: minerId,
mac: mac,
ip: ip,
model: model,
room: room,
rack: rack,
shelf: shelf,
position: position,
serial: serial,
location: location,
hashrate: hashrate,
temp: temp,
source: sourceName,
haystack: haystack,
normalized: normalize(haystack)
});
}
function addRowsFromArray(rows, arrayValue, sourceName) {
if (!Array.isArray(arrayValue)) {
return;
}
for (const item of arrayValue) {
addSearchRow(rows, item, sourceName);
}
}
function collectRowsFromPayload(rows, payload, sourceName) {
if (!payload) {
return;
}
if (Array.isArray(payload)) {
addRowsFromArray(rows, payload, sourceName);
return;
}
if (typeof payload !== "object") {
return;
}
addRowsFromArray(rows, payload.positions, sourceName + ".positions");
addRowsFromArray(rows, payload.virtual_miners, sourceName + ".virtual_miners");
addRowsFromArray(rows, payload.miners, sourceName + ".miners");
addRowsFromArray(rows, payload.items, sourceName + ".items");
addRowsFromArray(rows, payload.results, sourceName + ".results");
addRowsFromArray(rows, payload.rows, sourceName + ".rows");
addRowsFromArray(rows, payload.data, sourceName + ".data");
addSearchRow(rows, payload, sourceName);
}
function dedupeRows(rows) {
const seen = new Set();
const output = [];
for (const row of rows) {
const key = row.miner_id || normalize(row.mac) || row.ip || row.serial;
if (!key) {
continue;
}
if (seen.has(key)) {
continue;
}
seen.add(key);
output.push(row);
}
return output;
}
async function loadSearchCache() {
if (searchCache) {
return searchCache;
}
const rows = [];
const attempts = [];
const urls = [
"/api/location/map",
"/api/miners",
"/api/dashboard/miner-table",
"/api/dashboard/hot-miners",
"/api/dashboard/unstable-miners",
"/api/dashboard/stale-miners"
];
for (const url of urls) {
try {
const payload = await fetchJson(url);
const beforeCount = rows.length;
collectRowsFromPayload(rows, payload, url);
attempts.push(url + ": +" + (rows.length - beforeCount));
} catch (error) {
attempts.push(url + ": " + error.message);
}
}
searchCache = dedupeRows(rows);
lastLoadSummary = "Loaded " + searchCache.length + " searchable miners. " + attempts.join(" | ");
return searchCache;
}
function resultScore(row, rawQuery, normalizedQuery) {
const q = rawQuery.toLowerCase();
if (String(row.mac || "").toLowerCase().includes(q)) {
return 100;
}
if (normalize(row.mac).includes(normalizedQuery)) {
return 98;
}
if (String(row.ip || "").toLowerCase().includes(q)) {
return 92;
}
if (String(row.serial || "").toLowerCase().includes(q)) {
return 88;
}
if (String(row.location || "").toLowerCase().includes(q)) {
return 75;
}
if (String(row.room || "").toLowerCase().includes(q)) {
return 72;
}
if (String(row.rack || "").toLowerCase().includes(q)) {
return 70;
}
if (String(row.model || "").toLowerCase().includes(q)) {
return 62;
}
if (row.normalized.includes(normalizedQuery)) {
return 50;
}
return 0;
}
function setResultsMessage(message) {
const resultsBox = document.getElementById("mineropsGlobalSearchResults");
if (!resultsBox) {
return;
}
resultsBox.classList.add("open");
resultsBox.innerHTML = `