From ccabc9e256e4e1ce2533a9f7074d99c32bd943ef Mon Sep 17 00:00:00 2001 From: Chris Burr Date: Mon, 13 Jul 2026 10:39:21 +0200 Subject: [PATCH] fix: Escape metadata values in DirectoryMetadata.__createMetaSelection __createMetaSelection interpolated user-supplied metadata operands directly into the SQL selection string (e.g. "{table}Value='{operand}'"), and the callers (__findSubdirByMeta, __checkDirsForMetadata) execute the resulting query with no bound args, so a crafted metadata query value was live SQL -> injection via findDirectoriesByMetadata. Escape every string operand through the DB _escapeString/_escapeValues helpers (which quote and escape the value) before inlining it, mirroring the FileMetadata.__createMetaSelection sibling which already does this. Numeric operands keep their existing safe %d/%f handling. This defect predates the SQL-parameterisation series. --- .../DirectoryMetadata/DirectoryMetadata.py | 35 +++++++++++++++---- 1 file changed, 28 insertions(+), 7 deletions(-) diff --git a/src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/DirectoryMetadata/DirectoryMetadata.py b/src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/DirectoryMetadata/DirectoryMetadata.py index 65e8bcbb6c8..faf40e536a3 100644 --- a/src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/DirectoryMetadata/DirectoryMetadata.py +++ b/src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/DirectoryMetadata/DirectoryMetadata.py @@ -513,28 +513,49 @@ def __createMetaSelection(self, value, table=""): elif isinstance(operand, float): selectList.append(f"{table}Value{operation}{operand:f}") else: - selectList.append(f"{table}Value{operation}'{operand}'") + result = self.db._escapeString(operand) + if not result["OK"]: + return result + selectList.append(f"{table}Value{operation}{result['Value']}") elif operation == "in" or operation == "=": if isinstance(operand, list): - vString = ",".join(["'" + str(x) + "'" for x in operand]) + result = self.db._escapeValues(operand) + if not result["OK"]: + return result + vString = ",".join(result["Value"]) selectList.append(f"{table}Value IN ({vString})") else: - selectList.append(f"{table}Value='{operand}'") + result = self.db._escapeString(operand) + if not result["OK"]: + return result + selectList.append(f"{table}Value={result['Value']}") elif operation == "nin" or operation == "!=": if isinstance(operand, list): - vString = ",".join(["'" + str(x) + "'" for x in operand]) + result = self.db._escapeValues(operand) + if not result["OK"]: + return result + vString = ",".join(result["Value"]) selectList.append(f"{table}Value NOT IN ({vString})") else: - selectList.append(f"{table}Value!='{operand}'") + result = self.db._escapeString(operand) + if not result["OK"]: + return result + selectList.append(f"{table}Value!={result['Value']}") selectString = " AND ".join(selectList) elif isinstance(value, list): - vString = ",".join(["'" + str(x) + "'" for x in value]) + result = self.db._escapeValues(value) + if not result["OK"]: + return result + vString = ",".join(result["Value"]) selectString = f"{table}Value in ({vString})" else: if value == "Any": selectString = "" else: - selectString = f"{table}Value='{value}' " + result = self.db._escapeString(value) + if not result["OK"]: + return result + selectString = f"{table}Value={result['Value']} " return S_OK(selectString)