fix(security): close backend merge blockers

app/api/pagination.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class PaginatedList(list[T], Generic[T]):
+    """A page of items carrying the total count from its data source."""
+
+    def __init__(self, items: Iterable[T], *, total: int) -> None:
+        super().__init__(items)
+        self.total = total

app/api/v1/endpoints/audit.py

@@ -10,6 +10,7 @@ from app.auth.metadata_dependencies import (
     get_current_metadata_admin,
     get_current_metadata_user,
 )
+from app.api.pagination import PaginatedList
 from app.core.audit import AuditAction, log_audit_event
 from app.domain.schemas.audit import AuditLogResponse
 from app.infra.db.metadb.database import get_metadata_session
@@ -46,7 +47,7 @@ async def get_audit_logs(
     _current_user=Depends(get_current_metadata_admin),
     audit_repo: AuditRepository = Depends(get_audit_repository),
 ) -> list[AuditLogResponse]:
-    return await audit_repo.get_logs(
+    items = await audit_repo.get_logs(
         user_id=user_id,
         project_id=project_id,
         action=action,
@@ -56,6 +57,15 @@ async def get_audit_logs(
         skip=skip,
         limit=limit,
     )
+    total = await audit_repo.get_log_count(
+        user_id=user_id,
+        project_id=project_id,
+        action=action,
+        resource_type=resource_type,
+        start_time=start_time,
+        end_time=end_time,
+    )
+    return PaginatedList(items, total=total)
 
 
 @router.get(
@@ -119,7 +129,7 @@ async def get_my_audit_logs(
     current_user=Depends(get_current_metadata_user),
     audit_repo: AuditRepository = Depends(get_audit_repository),
 ) -> list[AuditLogResponse]:
-    return await audit_repo.get_logs(
+    items = await audit_repo.get_logs(
         user_id=current_user.id,
         action=action,
         start_time=start_time,
@@ -127,3 +137,10 @@ async def get_my_audit_logs(
         skip=skip,
         limit=limit,
     )
+    total = await audit_repo.get_log_count(
+        user_id=current_user.id,
+        action=action,
+        start_time=start_time,
+        end_time=end_time,
+    )
+    return PaginatedList(items, total=total)

app/api/v1/rest_router.py

@@ -14,6 +14,7 @@ from pydantic import BaseModel, JsonValue, create_model
 from starlette.responses import Response
 
 from app.api.problem_details import ProblemDetails
+from app.api.pagination import PaginatedList
 from app.api.v1.router import api_router as handler_api_router
 from app.auth.metadata_dependencies import get_current_metadata_user
 from app.auth.project_dependencies import ProjectContext, get_project_context
@@ -41,8 +42,8 @@ _PUBLIC_PARAMETER_RENAMES = {
     "burst_ID": "burst_id",
     "drainage_node_ID": "drainage_node_id",
 }
-_MODEL_NAME_IS_NETWORK = {"RunSimulationManuallyByDate"}
-_MODEL_USERNAME_FROM_AUTH: set[str] = set()
+_MODEL_NAME_IS_NETWORK = {"RunSimulationManuallyByDate", "PressureSensorPlacement"}
+_MODEL_USERNAME_FROM_AUTH = {"PressureSensorPlacement"}
 
 
 def _clean_name(name: str) -> str:
@@ -230,9 +231,14 @@ def _with_pagination(endpoint):
         if not isinstance(result, list):
             return result
         if handler_handles_pagination:
+            if not isinstance(result, PaginatedList):
+                raise RuntimeError(
+                    f"Paginated handler {endpoint.__name__!r} must return "
+                    "PaginatedList with the real total"
+                )
             return Page(
                 items=result,
-                total=offset + len(result),
+                total=result.total,
                 limit=limit or len(result),
                 offset=offset,
             )

app/api/v1/router.py

@@ -53,6 +53,7 @@ from app.api.v1.endpoints.timeseries import (
 )
 from app.auth.permissions import (
     BURST_RUN,
+    ENVIRONMENT_MANAGE,
     OPTIMIZATION_RUN,
     RISK_RUN,
     SCADA_CLEAN,
@@ -88,6 +89,7 @@ simulation_access = Depends(
 
 webgis_view_access = Depends(require_permission(WEBGIS_VIEW))
 simulation_run_access = Depends(require_permission(SIMULATION_RUN))
+environment_manage_access = Depends(require_permission(ENVIRONMENT_MANAGE))
 burst_run_access = Depends(require_permission(BURST_RUN))
 risk_run_access = Depends(require_permission(RISK_RUN))
 optimization_run_access = Depends(require_permission(OPTIMIZATION_RUN))
@@ -169,7 +171,7 @@ api_router.include_router(
 api_router.include_router(
     cache.router,
     tags=["Cache"],
-    dependencies=[simulation_run_access],
+    dependencies=[environment_manage_access],
 )
 api_router.include_router(
     web_search.router,

app/core/audit.py

@@ -130,6 +130,7 @@ def sanitize_sensitive_data(data: dict) -> dict:
         "token",
         "api_key",
         "apikey",
+        "dsn",
         "credit_card",
         "ssn",
         "social_security",

app/infra/db/influxdb/info.py

@@ -1,6 +1,5 @@
-# influxdb数据库连接信息
-url = "http://127.0.0.1:8086"  # 替换为你的InfluxDB实例地址
-token = "kMPX2V5HsbzPpUT2B9HPBu1sTG1Emf-lPlT2UjxYnGAuocpXq_f_0lK4HHs-TbbKyjsZpICkMsyXG_V2D7P7yQ=="  # 替换为你的InfluxDB Token
-# _ENCODED_TOKEN = "eEdETTVSWnFSSkF1ekFHUy1vdFhVZEMyTkZkWTc1cUpBalJMcUFCNHA1V2NJSUFsSVVwT3BUOF95QTE2QU9IbUpXZXJ3UV8wOGd3Yjg0c3k0MmpuWlE9PQ=="
-# token = base64.b64decode(_ENCODED_TOKEN).decode("utf-8")
-org = "TJWATERORG"  # 替换为你的Organization名称
+from app.core.config import settings
+
+url = settings.INFLUXDB_URL
+token = settings.INFLUXDB_TOKEN
+org = settings.INFLUXDB_ORG

app/native/wndb/database.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping, Sequence
 from typing import Any
 from psycopg.rows import dict_row, Row
 from .connection import project_connection
@@ -82,27 +83,38 @@ class DbChangeSet:
         return DbChangeSet(redo_sql, undo_sql, redo_cs_s, undo_cs_s)
 
 
-def read(name: str, sql: str) -> Row:
+QueryParams = Sequence[Any] | Mapping[str, Any]
+
+
+def _execute(cur, sql: str, params: QueryParams | None = None):
+    return cur.execute(sql, params) if params is not None else cur.execute(sql)
+
+
+def read(name: str, sql: str, params: QueryParams | None = None) -> Row:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(sql)
+            _execute(cur, sql, params)
             row = cur.fetchone()
             if row == None:
                 raise Exception(sql)
             return row
 
 
-def read_all(name: str, sql: str) -> list[Row]:
+def read_all(
+    name: str, sql: str, params: QueryParams | None = None
+) -> list[Row]:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(sql)
+            _execute(cur, sql, params)
             return cur.fetchall()
 
 
-def try_read(name: str, sql: str) -> Row | None:
+def try_read(
+    name: str, sql: str, params: QueryParams | None = None
+) -> Row | None:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(sql)
+            _execute(cur, sql, params)
             return cur.fetchone()
 
 

app/native/wndb/s0_base.py

@@ -1,3 +1,4 @@
+from psycopg import sql
 from psycopg.rows import dict_row, Row
 from .connection import project_connection
 from .database import read
@@ -49,7 +50,12 @@ ELEMENT_TYPES : dict[str, int] = {
 def _get_from(name: str, id: str, base_type: str) -> Row | None:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(f"select * from {base_type} where id = '{id}'")
+            cur.execute(
+                sql.SQL("select * from {} where id = %s").format(
+                    sql.Identifier(base_type)
+                ),
+                (id,),
+            )
             return cur.fetchone()
 
 
@@ -243,11 +249,17 @@ def get_node_links(name: str, id: str) -> list[str]:
     with project_connection(name) as conn:
         with conn.cursor(row_factory=dict_row) as cur:
             links: list[str] = []
-            for p in cur.execute(f"select id from pipes where node1 = '{id}' or node2 = '{id}'").fetchall():
+            for p in cur.execute(
+                "select id from pipes where node1 = %s or node2 = %s", (id, id)
+            ).fetchall():
                 links.append(p['id'])
-            for p in cur.execute(f"select id from pumps where node1 = '{id}' or node2 = '{id}'").fetchall():
+            for p in cur.execute(
+                "select id from pumps where node1 = %s or node2 = %s", (id, id)
+            ).fetchall():
                 links.append(p['id'])
-            for p in cur.execute(f"select id from valves where node1 = '{id}' or node2 = '{id}'").fetchall():
+            for p in cur.execute(
+                "select id from valves where node1 = %s or node2 = %s", (id, id)
+            ).fetchall():
                 links.append(p['id'])
             return links
 
@@ -255,16 +267,15 @@ def get_node_links(name: str, id: str) -> list[str]:
 def get_link_nodes(name: str, id: str) -> list[str]:
     row = {}
     if is_pipe(name, id):
-        row = read(name, f"select node1, node2 from pipes where id = '{id}'")
+        row = read(name, "select node1, node2 from pipes where id = %s", (id,))
     elif is_pump(name, id):
-        row = read(name, f"select node1, node2 from pumps where id = '{id}'")
+        row = read(name, "select node1, node2 from pumps where id = %s", (id,))
     elif is_valve(name, id):
-        row = read(name, f"select node1, node2 from valves where id = '{id}'")
+        row = read(name, "select node1, node2 from valves where id = %s", (id,))
     return [str(row['node1']), str(row['node2'])]
 
 def get_region_type(name: str, id: str)->str:
     if(is_region(name,id)):
-        type = read(name, f"select type from _region where id = '{id}'")
+        type = read(name, "select type from _region where id = %s", (id,))
         return type
 
-

app/native/wndb/s24_coordinates.py

@@ -23,7 +23,11 @@ def from_postgis_point(coord: str) -> dict[str, float]:
 
 
 def get_node_coord(name: str, node: str) -> dict[str, float]:
-    row = try_read(name, f"select st_astext(coord) as coord_geom from coordinates where node = '{node}'")
+    row = try_read(
+        name,
+        "select st_astext(coord) as coord_geom from coordinates where node = %s",
+        (node,),
+    )
     if row == None:
         write(name, sql_insert_coord(node, 0.0, 0.0))
         return {'x': 0.0, 'y': 0.0}
@@ -66,7 +70,9 @@ def get_links_in_extent(name: str, x1: float, y1: float, x2: float, y2: float) -
 
 
 def node_has_coord(name: str, node: str) -> bool:
-    return try_read(name, f"select node from coordinates where node = '{node}'") != None
+    return try_read(
+        name, "select node from coordinates where node = %s", (node,)
+    ) != None
 
 
 #--------------------------------------------------------------

app/native/wndb/s2_junctions.py

@@ -12,7 +12,7 @@ def get_junction_schema(name: str) -> dict[str, dict[str, Any]]:
 
 
 def get_junction(name: str, id: str) -> dict[str, Any]:
-    j = try_read(name, f"select * from junctions where id = '{id}'")
+    j = try_read(name, "select * from junctions where id = %s", (id,))
     if j == None:
         return {}
     xy = get_node_coord(name, id)