diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9bec228..86a9c96 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -202,6 +202,12 @@ follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   subqueries, or `FETCH`; anything but a non-negative integer literal within
   policy is replaced.
 - `SELECT ... INTO` is rejected as a non-read-only operation.
+- Table references that look like a file path or URL (DuckDB's
+  `FROM 'data.csv'` or `FROM 's3://b/x.parquet'`) are rejected. Previously they
+  were caught only as an unknown table, so a schema table named `csv` or
+  `parquet` let the file read through.
+- MySQL/MariaDB executable comments (`/*! ... */`, `/*M! ... */`) are rejected,
+  since the server runs their contents as SQL.
 - SQLite databases are opened read-only (`mode=ro`), matching the Postgres
   `READ ONLY` transaction, so a validator gap cannot become a write.
 
diff --git a/src/querypilot/validation/validator.py b/src/querypilot/validation/validator.py
index 1d7e474..38e9c80 100644
--- a/src/querypilot/validation/validator.py
+++ b/src/querypilot/validation/validator.py
@@ -159,6 +159,31 @@ def validate(self, sql: str, schema: DatabaseSchema) -> ValidationResult:
                 )
             )
 
+        source_error = _executable_comment_error(stripped, self.config.dialect) or (
+            _external_source_error(expression)
+        )
+        if source_error is not None:
+            readonly = False
+            errors.append(source_error)
+            blocked_reason = blocked_reason or source_error
+            risk_level = _max_risk(risk_level, "critical")
+            checks.append(
+                PolicyCheck(
+                    name="source_safety",
+                    passed=False,
+                    message=source_error,
+                    severity="critical",
+                )
+            )
+        else:
+            checks.append(
+                PolicyCheck(
+                    name="source_safety",
+                    passed=True,
+                    message="No executable comments or file and URL table references.",
+                )
+            )
+
         analysis = _QueryAnalysis()
         if is_query and non_readonly_operation is None:
             try:
@@ -393,6 +418,26 @@ def _function_policy_error(
     return None
 
 
+def _executable_comment_error(sql: str, dialect: str) -> str | None:
+    # MySQL runs the body of /*! ... */ (MariaDB also /*M! ... */) as SQL, so
+    # anything hidden there escapes validation.
+    for token in sqlglot.tokenize(sql, read=dialect):
+        for comment in token.comments:
+            if comment.startswith(("!", "M!")):
+                return "Executable comments (/*! ... */) are not allowed."
+    return None
+
+
+def _external_source_error(expression: exp.Expression) -> str | None:
+    # DuckDB reads `FROM 'data.csv'` or `FROM 's3://b/x.parquet'` as a file or URL
+    # (a replacement scan), so a table name that looks like a path is not a table.
+    for table in expression.find_all(exp.Table):
+        for part in table.parts:
+            if isinstance(part, exp.Identifier) and any(char in part.name for char in "./\\:"):
+                return f"Table reference looks like a file path or URL: {part.name}"
+    return None
+
+
 def _function_name(function: exp.Func) -> str:
     if isinstance(function, exp.Anonymous):
         return function.name.lower()
diff --git a/tests/test_validation.py b/tests/test_validation.py
index 2fca1db..92f8e8a 100644
--- a/tests/test_validation.py
+++ b/tests/test_validation.py
@@ -4,7 +4,7 @@
 
 from querypilot import QueryPilot
 from querypilot.core.config import QueryPilotConfig, SafetyPolicy
-from querypilot.core.types import DatabaseSchema
+from querypilot.core.types import ColumnSchema, DatabaseSchema, TableSchema
 from querypilot.validation.validator import SQLValidator
 
 
@@ -201,6 +201,90 @@ def test_dangerous_postgres_functions_are_rejected(function_name: str) -> None:
     assert result.blocked_reason == f"SQL function is blocked by policy: {function_name}"
 
 
+def _schema_with_tables(dialect: str, *names: str) -> DatabaseSchema:
+    return DatabaseSchema(
+        dialect=dialect,
+        tables=[
+            TableSchema(name=name, columns=[ColumnSchema(name="id", type="INTEGER")])
+            for name in names
+        ],
+    )
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT * FROM 'data.csv'",
+        "SELECT * FROM \"data.csv\"",
+        "SELECT p.id FROM 's3://bucket/x.parquet' AS p",
+        "SELECT * FROM 'https://example.com/x.json'",
+        "SELECT * FROM customers JOIN '/etc/passwd.csv' AS f ON f.id = customers.id",
+    ],
+)
+def test_file_and_url_table_references_are_rejected(sql: str) -> None:
+    # A schema table named like the file's extension must not let the read through.
+    validator = SQLValidator(QueryPilotConfig(dialect="duckdb"))
+    schema = _schema_with_tables("duckdb", "csv", "parquet", "json", "customers")
+
+    result = validator.validate(sql, schema)
+
+    assert result.valid is False
+    assert result.readonly is False
+    assert result.blocked_reason.startswith("Table reference looks like a file path or URL:")
+
+
+def test_schema_qualified_table_is_not_mistaken_for_a_path() -> None:
+    validator = SQLValidator(QueryPilotConfig(dialect="duckdb"))
+
+    result = validator.validate(
+        "SELECT id FROM main.customers", _schema_with_tables("duckdb", "customers")
+    )
+
+    assert result.valid is True
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT id /*!50000 , SLEEP(5) */ FROM t",
+        "SELECT id FROM t /*!UNION SELECT password FROM users */",
+        "SELECT id /*M!100000 , SLEEP(5) */ FROM t",
+    ],
+)
+def test_mysql_executable_comments_are_rejected(sql: str) -> None:
+    validator = SQLValidator(QueryPilotConfig(dialect="mysql"))
+
+    result = validator.validate(sql, _schema_with_tables("mysql", "t"))
+
+    assert result.valid is False
+    assert result.blocked_reason == "Executable comments (/*! ... */) are not allowed."
+
+
+def test_ordinary_and_hint_comments_are_allowed() -> None:
+    validator = SQLValidator(QueryPilotConfig(dialect="mysql"))
+    schema = _schema_with_tables("mysql", "t")
+
+    assert validator.validate("SELECT id /* plain note */ FROM t", schema).valid is True
+    assert validator.validate("SELECT /*+ MAX_EXECUTION_TIME(1000) */ id FROM t", schema).valid
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT * FROM OPENROWSET(BULK 'C:\\secrets.txt', SINGLE_CLOB) AS x",
+        "SELECT * FROM OPENROWSET('SQLNCLI', 'Server=evil;', 'SELECT 1') AS x",
+        "SELECT * FROM OPENQUERY(linked, 'SELECT 1')",
+        "SELECT * FROM OPENDATASOURCE('SQLNCLI', 'Data Source=evil').db.dbo.t",
+    ],
+)
+def test_tsql_external_rowsets_are_rejected(sql: str) -> None:
+    validator = SQLValidator(QueryPilotConfig(dialect="tsql"))
+
+    result = validator.validate(sql, _schema_with_tables("tsql", "t", "x"))
+
+    assert result.valid is False
+
+
 def test_function_allowlist_fails_closed() -> None:
     validator = SQLValidator(
         QueryPilotConfig(