71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_POLICY = ROOT / "plugins" / "1c" / "connector" / "policies" / "read-only-query.yaml"
|
|
|
|
|
|
def load_policy(path: Path) -> dict:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
data = yaml.safe_load(handle)
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} must contain a YAML mapping")
|
|
return data
|
|
|
|
|
|
def validate_query(query: str, policy: dict) -> list[str]:
|
|
errors = []
|
|
normalized = query.strip()
|
|
if not normalized:
|
|
return ["query is empty"]
|
|
if not re.match(r"(?is)^\s*(ВЫБРАТЬ|SELECT)\b", normalized):
|
|
errors.append("only SELECT/ВЫБРАТЬ queries are allowed")
|
|
for pattern in policy.get("deny_patterns") or []:
|
|
if re.search(pattern, normalized):
|
|
errors.append(f"query matches denied pattern: {pattern}")
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Validate a read-only 1C query request.")
|
|
parser.add_argument("--query")
|
|
parser.add_argument("--request-json", type=Path)
|
|
parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY)
|
|
parser.add_argument("--json", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
if args.request_json:
|
|
request = json.loads(args.request_json.read_text(encoding="utf-8"))
|
|
query = request.get("query") or ""
|
|
else:
|
|
query = args.query or ""
|
|
|
|
policy = load_policy(args.policy)
|
|
errors = validate_query(query, policy)
|
|
result = {
|
|
"allowed": not errors,
|
|
"errors": errors,
|
|
"limits": policy.get("limits") or {},
|
|
}
|
|
if args.json:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
elif errors:
|
|
print("Query is denied:", file=sys.stderr)
|
|
for error in errors:
|
|
print(f"- {error}", file=sys.stderr)
|
|
else:
|
|
print("Query is allowed.")
|
|
return 0 if not errors else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|