Visitar URL original
fix(spanner): harden identifier escaping in dbapi driver by olavloite · Pull Request #19054 · googleapis/google-cloud-python · GitHub
Skip to content

fix(spanner): harden identifier escaping in dbapi driver #19054

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
olavloite wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(spanner): harden identifier escaping in dbapi driver #19054

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 24 additions & 18 deletions packages/google-cloud-spanner/google/cloud/spanner_dbapi/parse_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,9 @@
JsonObject: spanner.param_types.JSON,
}

# Note: SPANNER_RESERVED_KEYWORDS is retained for backwards compatibility with
# downstream packages (e.g. sqlalchemy-spanner). The Spanner DB-API driver quotes
# identifiers unconditionally and does not use this list internally.
SPANNER_RESERVED_KEYWORDS = {
"ALL",
"AND",
Expand Down Expand Up @@ -89,6 +92,7 @@
"FOR",
"FROM",
"FULL",
"GRAPH_TABLE",
"GROUP",
"GROUPING",
"GROUPS",
Expand Down Expand Up @@ -124,6 +128,7 @@
"PARTITION",
"PRECEDING",
"PROTO",
"QUALIFY",
"RANGE",
"RECURSIVE",
"RESPECT",
Expand Down Expand Up @@ -383,37 +388,38 @@ def ensure_where_clause(sql):


_QUOTED_IDENTIFIER_RE = re.compile(r"\A`(\\.|[^`\\])*`\Z", re.DOTALL)
_QUALIFIED_PART_RE = re.compile(r"`(?:\\.|[^`\\])*`|[^.]+")
_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f]")


def escape_name(name):
"""
Apply backticks to the name if it is not a valid regular ASCII identifier,
or if it is a Cloud Spanner's reserved keyword.
def escape_name(name: str) -> str:
"""Apply GoogleSQL backticks to the identifier and escape embedded delimiters.

:type name: str
:param name: Name to escape.
:param name: Identifier name to escape.

:rtype: str
:returns: Name escaped if it has to be escaped.
:returns: Escaped and quoted identifier.
:raises TypeError: if the name is not a string.
:raises ValueError: if the name contains ASCII control characters or is
an invalid qualified identifier.
"""
if not isinstance(name, str):
raise TypeError(f"Identifier must be a string, got {type(name).__name__}")
if not name:
return name

if _CONTROL_CHAR_RE.search(name):
raise ValueError(f"Invalid identifier: {name!r}")

if _QUOTED_IDENTIFIER_RE.match(name):
return name

if "." in name:
parts = name.split(".")
parts = _QUALIFIED_PART_RE.findall(name)
if not parts or ".".join(parts) != name:
raise ValueError(f"Invalid qualified identifier: {name!r}")
return ".".join(escape_name(part) for part in parts)

is_valid_regular_identifier = (
(name[0].isalpha() or name[0] == "_")
and all(c.isalnum() or c == "_" for c in name)
and name.isascii()
)

if not is_valid_regular_identifier or name.upper() in SPANNER_RESERVED_KEYWORDS:
escaped = name.replace("\\", "\\\\").replace("`", "\\`")
return f"`{escaped}`"

return name
escaped = name.replace("\\", "\\\\").replace("`", "\\`")
return f"`{escaped}`"
70 changes: 64 additions & 6 deletions packages/google-cloud-spanner/tests/unit/spanner_dbapi/test_parse_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,11 @@
import sys
import unittest

from google.cloud.spanner_dbapi.parse_utils import classify_statement
from google.cloud.spanner_dbapi.parse_utils import (
SPANNER_RESERVED_KEYWORDS,
classify_statement,
escape_name,
)
from google.cloud.spanner_dbapi.parsed_statement import (
ClientSideStatementType,
ParsedStatement,
Expand Down Expand Up @@ -501,27 +505,32 @@ def test_ensure_where_clause(self):

@unittest.skipIf(skip_condition, skip_message)
def test_escape_name(self):
from google.cloud.spanner_dbapi.parse_utils import escape_name

cases = (
("SELECT", "`SELECT`"),
("dashed-value", "`dashed-value`"),
("with space", "`with space`"),
("name", "name"),
("name", "`name`"),
("", ""),
("``", "``"),
("col`; DROP TABLE t; -- x", "`col\\`; DROP TABLE t; -- x`"),
("table`name", "`table\\`name`"),
("`", "`\\``"),
("col/*comment*/name", "`col/*comment*/name`"),
("123column", "`123column`"),
("col;select", "`col;select`"),
("col\nname", "`col\nname`"),
("test\\", "`test\\\\`"),
("my_schema.my_table", "my_schema.my_table"),
("my_schema.my_table", "`my_schema`.`my_table`"),
("my-schema.my-table", "`my-schema`.`my-table`"),
("catalog.schema.table.column", "`catalog`.`schema`.`table`.`column`"),
("`my_table`", "`my_table`"),
("`my_schema`.`my_table`", "`my_schema`.`my_table`"),
("`table.with.dots`", "`table.with.dots`"),
("`table.with.dots`.col", "`table.with.dots`.`col`"),
("col.`table.with.dots`", "`col`.`table.with.dots`"),
("`my.schema`.`my.table`", "`my.schema`.`my.table`"),
("café", "`café`"),
("col_ñ", "`col_ñ`"),
("`col\\`name`", "`col\\`name`"),
("`col\\`; DROP TABLE users; --`", "`col\\`; DROP TABLE users; --`"),
(
"`col\\\\`; DROP TABLE users; --`",
Expand All @@ -532,3 +541,52 @@ def test_escape_name(self):
with self.subTest(name=name):
got = escape_name(name)
self.assertEqual(got, want)
# Verify idempotency: escaping an already-escaped output returns the same result.
self.assertEqual(escape_name(got), want)

@unittest.skipIf(skip_condition, skip_message)
def test_escape_name_invalid_control_characters(self):
invalid_identifiers = (
"table\nname",
"table\rname",
"table\0name",
"table\tname",
"table\x1fname",
"table\x7fname",
"schema\n.table",
"schema.table\0",
)
for identifier in invalid_identifiers:
with self.subTest(identifier=identifier):
with self.assertRaises(ValueError):
escape_name(identifier)

@unittest.skipIf(skip_condition, skip_message)
def test_escape_name_invalid_qualified_identifiers(self):
invalid_qualified = (
".",
"...",
"table.",
".table",
"schema..table",
"schema.`unclosed.table.`other",
)
for identifier in invalid_qualified:
with self.subTest(identifier=identifier):
with self.assertRaises(ValueError):
escape_name(identifier)

@unittest.skipIf(skip_condition, skip_message)
def test_escape_name_invalid_types(self):
invalid_types = (None, 123, 45.67, b"bytes_name", ["list"])
for identifier in invalid_types:
with self.subTest(identifier=identifier):
with self.assertRaises(TypeError):
escape_name(identifier)

@unittest.skipIf(skip_condition, skip_message)
def test_spanner_reserved_keywords_compatibility(self):
# Verify that SPANNER_RESERVED_KEYWORDS is exported with recent additions
# for backwards compatibility with downstream consumers (e.g. sqlalchemy-spanner).
self.assertIn("GRAPH_TABLE", SPANNER_RESERVED_KEYWORDS)
self.assertIn("QUALIFY", SPANNER_RESERVED_KEYWORDS)
Loading