commit 47b8f5f74613ff31f058e7238d5c26dbbfc68c87 Author: Bart Broere Date: Wed Oct 14 10:46:20 2026 +0200 Add peo4c: neo4j-browser on PostgreSQL + Apache AGE A single container image with: - neo4j-browser built from source and served on :7474 - PostgreSQL with Apache AGE (based on apache/age) - a Python bridge speaking Bolt 4.4/5.0-5.4 over TCP and WebSocket on :7687 that runs Cypher through AGE's cypher() function, emulates the Neo4j procedures neo4j-browser relies on, and maps databases to AGE graphs Includes workarounds for AGE limitations (parameters are inlined as literals, `id(x) IN [...]` and the browser's expand-node query are rewritten) and an end-to-end smoke test using the Neo4j Python driver. Co-Authored-By: Claude Opus 5.5 diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..6e75e67 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,5 @@ +.git +**/__pycache__ +**/*.pyc +.venv +tests diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..00f2d38 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.venv/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..bad42ea --- /dev/null +++ b/Dockerfile @@ -0,0 +1,46 @@ +# peo4c: neo4j-browser on top of PostgreSQL + Apache AGE. +# +# Stage 1 builds the neo4j-browser static frontend from source. +# Stage 2 is the Apache AGE image (PostgreSQL with the AGE extension) plus the +# peo4c bridge, which speaks the Bolt protocol to the browser and translates +# Cypher into AGE `cypher()` calls. + +ARG AGE_IMAGE=docker.io/apache/age:release_PG18_1.8.0 + +FROM docker.io/library/node:16.20.2-bullseye AS browser + +ARG NEO4J_BROWSER_REPO=https://github.com/neo4j/neo4j-browser.git +ARG NEO4J_BROWSER_REF=3928c10fd6f148fd615b391d27d3fd6ce43a9044 + +WORKDIR /src +RUN git init -q . \ + && git remote add origin "$NEO4J_BROWSER_REPO" \ + && git fetch -q --depth 1 origin "$NEO4J_BROWSER_REF" \ + && git checkout -q FETCH_HEAD +RUN yarn install --frozen-lockfile --network-timeout 600000 +RUN yarn build && test -f dist/index.html + + +FROM ${AGE_IMAGE} + +RUN apt-get update \ + && apt-get install -y --no-install-recommends python3 python3-psycopg \ + && rm -rf /var/lib/apt/lists/* + +COPY --from=browser /src/dist /opt/peo4c/browser +COPY bridge /opt/peo4c/bridge +COPY docker/initdb/ /docker-entrypoint-initdb.d/ +COPY docker/peo4c-entrypoint.sh /usr/local/bin/peo4c-entrypoint.sh + +ENV PEO4C_DEFAULT_GRAPH=neo4j \ + PEO4C_BROWSER_DIR=/opt/peo4c/browser \ + PEO4C_HTTP_PORT=7474 \ + PEO4C_BOLT_PORT=7687 \ + PEO4C_PG_HOST=127.0.0.1 \ + PEO4C_PG_PORT=5432 + +# 7474: neo4j-browser (HTTP), 7687: Bolt, 5432: PostgreSQL +EXPOSE 7474 7687 5432 + +ENTRYPOINT ["/usr/local/bin/peo4c-entrypoint.sh"] +CMD ["postgres", "-c", "shared_preload_libraries=age"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..de05a93 --- /dev/null +++ b/README.md @@ -0,0 +1,99 @@ +# peo4c + +[neo4j-browser](https://github.com/neo4j/neo4j-browser) running on PostgreSQL, using +[Apache AGE](https://github.com/apache/age) as the graph engine. + +Everything ships in one container image: + +``` + neo4j-browser (static files) peo4c bridge (Python) PostgreSQL + AGE + ┌──────────────────────┐ Bolt ┌──────────────────────────┐ SQL ┌────────────────┐ + │ in your web browser │───────▶│ Bolt server (TCP + WS) │──────▶│ cypher('g',$$ │ + │ served on :7474 │ :7687 │ Cypher → AGE translation │ :5432 │ ... $$) │ + └──────────────────────┘ │ emulated Neo4j procedures │ └────────────────┘ + └──────────────────────────┘ +``` + +## Quick start + +```sh +podman build -t peo4c . +podman run -d --name peo4c -e POSTGRES_PASSWORD=secret \ + -p 7474:7474 -p 7687:7687 -p 5432:5432 \ + -v peo4c-data:/var/lib/postgresql peo4c +``` + +Open and log in with your **PostgreSQL** credentials: +user `postgres`, password the value of `POSTGRES_PASSWORD`. Bolt clients (the Neo4j +drivers, cypher-shell) can connect to `neo4j://localhost:7687` the same way, and +`psql` works on port 5432. + +`docker` works as a drop-in replacement for `podman` in the commands above. + +## How it works + +- **Databases are AGE graphs.** Every AGE graph shows up as a Neo4j database, and + `:use ` switches between them. A graph named `neo4j` is created on first start + (`PEO4C_DEFAULT_GRAPH`). `CREATE DATABASE x` and `DROP DATABASE x` on the `system` + database call `create_graph` / `drop_graph`; AGE requires names of at least 3 characters. +- **Queries** are sent to AGE as `SELECT * FROM cypher('', $$ $$) AS (...)`. + The bridge works out the result columns from the final `RETURN` clause and inlines + query parameters as Cypher literals. +- **Neo4j procedures** that neo4j-browser relies on (`dbms.components`, `SHOW DATABASES`, + `db.labels`, `db.schema.visualization`, `SHOW PROCEDURES`, ...) are emulated from the + AGE catalog. +- **Workarounds for AGE**: neo4j-browser's "expand node" query uses a pattern + comprehension, and its relationship lookup uses `id(n) IN [...]`, which AGE rejects + for lists of more than one element. The bridge rewrites both into equivalent Cypher + that AGE accepts. +- **Authentication** is passed straight through: every Bolt connection opens its own + PostgreSQL connection with the credentials you log in with. + +## Configuration + +| Variable | Default | Purpose | +| --- | --- | --- | +| `POSTGRES_PASSWORD` | (required) | Password of the PostgreSQL superuser, see the [postgres image docs](https://hub.docker.com/_/postgres). `POSTGRES_USER` and `POSTGRES_DB` work too. | +| `PEO4C_DEFAULT_GRAPH` | `neo4j` | Graph created at initialisation and used when a client does not pick a database. | +| `PEO4C_PG_DATABASE` | `$POSTGRES_DB` | PostgreSQL database holding the graphs. | +| `PEO4C_ADVERTISED_BOLT_PORT` | `7687` | Bolt port advertised to the browser; set it when you publish 7687 on another host port. | +| `PEO4C_ADVERTISED_HOST` | `localhost` | Host reported in `SHOW DATABASES` and to clients that do not send a routing address. | +| `PEO4C_LOG_LEVEL` | `INFO` | `DEBUG` logs every generated SQL statement. | + +Build arguments: `AGE_IMAGE` (base image, default `apache/age:release_PG18_1.8.0`) and +`NEO4J_BROWSER_REF` (neo4j-browser commit to build). + +## Limitations + +This is a first version. Things that do not work (yet): + +- `RETURN *`, `EXPLAIN` and `PROFILE`. +- Write statistics: write queries report "(no changes, no records)" even when they + changed the graph. +- Anything AGE does not support, for example multiple labels per node, pattern + comprehensions, temporal and spatial types, indexes and constraints through Cypher. +- User and role management, `SHOW TRANSACTIONS`, `:sysinfo`. +- TLS (`bolt+s://`, HTTPS). +- Each query's full result is fetched from PostgreSQL before it is streamed to the + client, so very large results use memory in the bridge. + +## Development + +The bridge is a dependency-light Python package in [`bridge/peo4c`](bridge/peo4c) +(only `psycopg` 3). To run it outside the container against any AGE-enabled +PostgreSQL: + +```sh +pip install 'psycopg[binary]' +cd bridge +PEO4C_PG_HOST=localhost PEO4C_PG_DATABASE=postgres PEO4C_BROWSER_DIR=/path/to/neo4j-browser/dist \ + python -m peo4c +``` + +[`tests/smoke_test.py`](tests/smoke_test.py) runs end-to-end checks through the official +Neo4j Python driver against a running instance: + +```sh +pip install neo4j +python tests/smoke_test.py neo4j://localhost:7687 postgres secret +``` diff --git a/bridge/peo4c/__init__.py b/bridge/peo4c/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/bridge/peo4c/__main__.py b/bridge/peo4c/__main__.py new file mode 100644 index 0000000..2c59a44 --- /dev/null +++ b/bridge/peo4c/__main__.py @@ -0,0 +1,51 @@ +"""peo4c bridge: serves neo4j-browser and translates Bolt/Cypher to PostgreSQL + AGE.""" + +import asyncio +import functools +import logging +import os + +from .backend import Config +from .bolt import handle_client +from .web import HttpServer + + +def env(name, default): + return os.environ.get(name) or default + + +async def main(): + logging.basicConfig(level=env("PEO4C_LOG_LEVEL", "INFO").upper(), + format="%(asctime)s %(levelname)s %(name)s: %(message)s") + log = logging.getLogger("peo4c") + + bolt_port = int(env("PEO4C_BOLT_PORT", "7687")) + http_port = int(env("PEO4C_HTTP_PORT", "7474")) + advertised_port = env("PEO4C_ADVERTISED_BOLT_PORT", str(bolt_port)) + config = Config( + pg_host=env("PEO4C_PG_HOST", "127.0.0.1"), + pg_port=int(env("PEO4C_PG_PORT", "5432")), + pg_database=env("PEO4C_PG_DATABASE", env("POSTGRES_DB", env("POSTGRES_USER", "postgres"))), + default_graph=env("PEO4C_DEFAULT_GRAPH", "neo4j"), + advertised_bolt=f"{env('PEO4C_ADVERTISED_HOST', 'localhost')}:{advertised_port}", + ) + + bolt = await asyncio.start_server(functools.partial(handle_client, config=config), "0.0.0.0", bolt_port) + servers = [bolt] + log.info("Bolt listening on :%d (PostgreSQL %s:%d/%s, default graph %r)", bolt_port, + config.pg_host, config.pg_port, config.pg_database, config.default_graph) + + browser_dir = env("PEO4C_BROWSER_DIR", "") + if browser_dir: + http = HttpServer(browser_dir, bolt_port) + servers.append(await asyncio.start_server(http.handle, "0.0.0.0", http_port)) + log.info("neo4j-browser on http://localhost:%d/browser/", http_port) + + await asyncio.gather(*(s.serve_forever() for s in servers)) + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + pass diff --git a/bridge/peo4c/agtype.py b/bridge/peo4c/agtype.py new file mode 100644 index 0000000..f9baf2c --- /dev/null +++ b/bridge/peo4c/agtype.py @@ -0,0 +1,149 @@ +"""Parser for the text representation of AGE's ``agtype``. + +agtype text is JSON extended with ``NaN``/``Infinity`` and type annotations +such as ``{...}::vertex``, ``{...}::edge``, ``[...]::path`` and ``1.5::numeric``. +""" + +import json + +from .packstream import Node, Path, Relationship + +# Labels AGE uses for vertices/edges created without an explicit label. +DEFAULT_LABELS = ("", "_ag_label_vertex", "_ag_label_edge") + +_SPECIAL = {"true": True, "false": False, "null": None, + "NaN": float("nan"), "Infinity": float("inf"), "-Infinity": float("-inf")} + + +class AgtypeError(ValueError): + pass + + +def parse(text): + if text is None: + return None + parser = _Parser(text) + value = parser.value() + parser.ws() + if parser.pos != len(text): + raise AgtypeError(f"trailing data in agtype value: {text!r}") + return value + + +class _Parser: + def __init__(self, text): + self.s = text + self.pos = 0 + + def ws(self): + s, n = self.s, len(self.s) + while self.pos < n and s[self.pos] in " \t\r\n": + self.pos += 1 + + def value(self): + self.ws() + if self.pos >= len(self.s): + raise AgtypeError("unexpected end of agtype value") + c = self.s[self.pos] + if c == "{": + v = self.obj() + elif c == "[": + v = self.arr() + elif c == '"': + v = self.string() + else: + v = self.scalar() + return self.annotate(v) + + def annotate(self, v): + if not self.s.startswith("::", self.pos): + return v + self.pos += 2 + start = self.pos + while self.pos < len(self.s) and (self.s[self.pos].isalnum() or self.s[self.pos] == "_"): + self.pos += 1 + kind = self.s[start:self.pos] + if kind == "vertex": + label = v.get("label", "") + return Node(v["id"], [] if label in DEFAULT_LABELS else [label], v.get("properties") or {}) + if kind == "edge": + return Relationship(v["id"], v["start_id"], v["end_id"], v.get("label", ""), + v.get("properties") or {}) + if kind == "path": + return Path(v[0::2], v[1::2]) + if kind == "numeric": + return float(v) + return v + + def obj(self): + self.pos += 1 + result = {} + self.ws() + if self.s[self.pos] == "}": + self.pos += 1 + return result + while True: + self.ws() + key = self.string() + self.ws() + self.expect(":") + result[key] = self.value() + self.ws() + if self.s[self.pos] == ",": + self.pos += 1 + continue + self.expect("}") + return result + + def arr(self): + self.pos += 1 + result = [] + self.ws() + if self.s[self.pos] == "]": + self.pos += 1 + return result + while True: + result.append(self.value()) + self.ws() + if self.s[self.pos] == ",": + self.pos += 1 + continue + self.expect("]") + return result + + def string(self): + if self.s[self.pos] != '"': + raise AgtypeError(f"expected string at {self.pos} in {self.s!r}") + i = self.pos + 1 + n = len(self.s) + while i < n: + c = self.s[i] + if c == "\\": + i += 2 + continue + if c == '"': + break + i += 1 + raw = self.s[self.pos:i + 1] + self.pos = i + 1 + return json.loads(raw) + + def scalar(self): + start = self.pos + n = len(self.s) + while self.pos < n and self.s[self.pos] not in ",]}: \t\r\n": + self.pos += 1 + token = self.s[start:self.pos] + if token in _SPECIAL: + return _SPECIAL[token] + try: + if any(ch in token for ch in ".eE") and not token.lstrip("-").isdigit(): + return float(token) + return int(token) + except ValueError: + raise AgtypeError(f"invalid agtype token {token!r}") from None + + def expect(self, ch): + if self.pos >= len(self.s) or self.s[self.pos] != ch: + raise AgtypeError(f"expected {ch!r} at {self.pos} in {self.s!r}") + self.pos += 1 diff --git a/bridge/peo4c/backend.py b/bridge/peo4c/backend.py new file mode 100644 index 0000000..132edb7 --- /dev/null +++ b/bridge/peo4c/backend.py @@ -0,0 +1,396 @@ +"""Executes Bolt queries against PostgreSQL + Apache AGE. + +Every Neo4j "database" maps to an AGE graph. The special ``system`` database +only answers the administrative queries that neo4j-browser issues. +""" + +import logging +import re +import secrets +import time +from dataclasses import dataclass, field + +import psycopg +from psycopg import errors as pg_errors +from psycopg import sql + +from . import agtype +from .cypher import (CypherError, expand_id_in_lists, first_keyword, inline_parameters, + return_columns, rewrite_for_age, strip_statement) +from .packstream import Node, Relationship + +log = logging.getLogger("peo4c.backend") + +NEO4J_VERSION = "5.26.0" +SYSTEM_DB = "system" +SYSTEM_LABELS = ("_ag_label_vertex", "_ag_label_edge") +GRAPH_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.\-]{2,62}$") + +# Procedures emulated by peo4c, reported through SHOW PROCEDURES. +PROCEDURES = [ + ("db.labels", "List all node labels in the graph.", "db.labels() :: (label :: STRING)"), + ("db.relationshipTypes", "List all relationship types in the graph.", + "db.relationshipTypes() :: (relationshipType :: STRING)"), + ("db.propertyKeys", "List property keys in the graph.", "db.propertyKeys() :: (propertyKey :: STRING)"), + ("db.schema.visualization", "Visualize the schema of the graph.", + "db.schema.visualization() :: (nodes :: LIST, relationships :: LIST)"), + ("dbms.components", "List DBMS components and their versions.", + "dbms.components() :: (name :: STRING, versions :: LIST, edition :: STRING)"), + ("dbms.clientConfig", "Return config settings interesting to clients.", + "dbms.clientConfig() :: (name :: STRING, value :: STRING)"), + ("dbms.showCurrentUser", "Show the current user.", + "dbms.showCurrentUser() :: (username :: STRING, roles :: LIST, flags :: LIST)"), +] + +CLIENT_CONFIG = { + "browser.allow_outgoing_connections": "true", + "browser.credential_timeout": "0s", + "browser.retain_connection_credentials": "true", + "browser.retain_editor_history": "true", + "client.allow_telemetry": "false", + "dbms.security.auth_enabled": "true", +} + + +@dataclass +class Config: + pg_host: str + pg_port: int + pg_database: str + default_graph: str + advertised_bolt: str + + +@dataclass +class Result: + fields: list + rows: list = field(default_factory=list) + db: str = "" + query_type: str = "r" + t_first: int = 0 + + +class BoltFailure(Exception): + def __init__(self, code, message): + super().__init__(message) + self.code = code + self.message = message + + +def failure_from_pg(e: Exception) -> BoltFailure: + """Map a PostgreSQL/AGE error onto a Neo4j status code.""" + sqlstate = getattr(e, "sqlstate", None) or "" + diag = getattr(e, "diag", None) + message = ((diag.message_primary if diag else None) or str(e)).strip() + if sqlstate == "42601": + code = "Neo.ClientError.Statement.SyntaxError" + elif sqlstate.startswith("42"): + code = "Neo.ClientError.Statement.SemanticError" + elif sqlstate.startswith("22"): + code = "Neo.ClientError.Statement.ArgumentError" + elif sqlstate.startswith("23"): + code = "Neo.ClientError.Schema.ConstraintValidationFailed" + elif sqlstate.startswith("28"): + code = "Neo.ClientError.Security.Unauthorized" + elif sqlstate.startswith("40"): + code = "Neo.TransientError.Transaction.DeadlockDetected" + elif sqlstate == "57014": + code = "Neo.ClientError.Transaction.TransactionTimedOut" + elif sqlstate.startswith("08") or isinstance(e, psycopg.OperationalError): + code = "Neo.TransientError.General.DatabaseUnavailable" + else: + code = "Neo.DatabaseError.General.UnknownError" + return BoltFailure(code, message) + + +def _norm(query: str) -> str: + return re.sub(r"\s+", " ", strip_statement(query)).strip() + + +class Session: + """One PostgreSQL connection, owned by one Bolt connection.""" + + def __init__(self, config: Config, user: str): + self.config = config + self.user = user + self.conn: psycopg.AsyncConnection | None = None + self.tx_db: str | None = None + + async def connect(self, password: str): + try: + self.conn = await psycopg.AsyncConnection.connect( + host=self.config.pg_host, port=self.config.pg_port, dbname=self.config.pg_database, + user=self.user, password=password, autocommit=True, + application_name="peo4c") + except psycopg.OperationalError as e: + msg = str(e) + if "password authentication failed" in msg or "role" in msg and "does not exist" in msg: + raise BoltFailure("Neo.ClientError.Security.Unauthorized", + "The client is unauthorized due to authentication failure.") from e + raise BoltFailure("Neo.TransientError.General.DatabaseUnavailable", + f"Could not connect to PostgreSQL: {msg.strip()}") from e + try: + await self.conn.execute("LOAD 'age'") + except psycopg.Error: + pass # Already preloaded, or not permitted for this role. + await self.conn.execute('SET search_path = ag_catalog, "$user", public') + + async def close(self): + if self.conn is not None: + await self.conn.close() + self.conn = None + + # -- transactions --------------------------------------------------------- + + @property + def in_transaction(self): + return self.tx_db is not None + + async def begin(self, db): + db = await self.resolve_db(db) + await self._sql("BEGIN") + self.tx_db = db + + async def commit(self): + self.tx_db = None + await self._sql("COMMIT") + + async def rollback(self): + self.tx_db = None + if self.conn is not None and not self.conn.closed: + try: + await self._sql("ROLLBACK") + except BoltFailure: + pass + + async def reset(self): + await self.rollback() + + # -- query execution ------------------------------------------------------ + + async def _sql(self, query, params=None): + try: + cur = await self.conn.execute(query, params) + except psycopg.Error as e: + raise failure_from_pg(e) from e + return await cur.fetchall() if cur.description else [] + + async def graphs(self): + rows = await self._sql("SELECT name FROM ag_catalog.ag_graph ORDER BY name") + return [r[0] for r in rows] + + async def resolve_db(self, db): + if not db: + db = self.config.default_graph + if db.lower() == SYSTEM_DB: + return SYSTEM_DB + for name in await self.graphs(): + if name == db or name.lower() == db.lower(): + return name + raise BoltFailure("Neo.ClientError.Database.DatabaseNotFound", + f"Database does not exist. Database name: '{db}'.") + + async def run(self, query: str, params: dict, db: str | None) -> Result: + started = time.monotonic() + db = self.tx_db if self.in_transaction else await self.resolve_db(db) + try: + result = await self._run_system(query, db) + if result is None: + if db == SYSTEM_DB: + raise BoltFailure("Neo.ClientError.Statement.NotSystemDatabaseCommand", + "This is an administration command and it should be executed " + "against the system database: peo4c only supports a few of them.") + result = await self._run_cypher(query, params or {}, db) + except CypherError as e: + raise BoltFailure(e.code, e.message) from e + result.db = db + result.t_first = int((time.monotonic() - started) * 1000) + return result + + async def _run_cypher(self, query, params, graph) -> Result: + text = strip_statement(query) + keyword = first_keyword(text) + if keyword in ("EXPLAIN", "PROFILE"): + raise CypherError("Neo.ClientError.Statement.FeatureNotSupported", + f"{keyword} is not supported by Apache AGE.") + text = expand_id_in_lists(inline_parameters(rewrite_for_age(text), params)) + columns = return_columns(text) + tag = "peo4c" + while f"${tag}$" in text: + tag = "peo4c_" + secrets.token_hex(4) + col_defs = ", ".join(f"c{i} ag_catalog.agtype" for i in range(max(len(columns), 1))) + stmt = (f"SELECT * FROM ag_catalog.cypher({sql.Literal(graph).as_string(self.conn)}, " + f"${tag}$ {text} ${tag}$) AS ({col_defs})") + log.debug("SQL: %s", stmt) + rows = await self._sql(stmt) + if columns: + rows = [[agtype.parse(v) for v in row] for row in rows] + else: + rows = [] + writes = re.search(r"(? Result | None: + q = _norm(query) + ql = q.lower() + + if ql.startswith("call dbms.components()"): + return Result(["name", "versions", "edition"], [["Neo4j Kernel", [NEO4J_VERSION], "community"]]) + + if ql.startswith("call dbms.clientconfig()") or ql.startswith("call dbms.listconfig()"): + return Result(["name", "value"], [[k, v] for k, v in CLIENT_CONFIG.items()]) + + if ql.startswith("call dbms.showcurrentuser()") or ql == "show current user": + return Result(["username", "roles", "flags"], [[self.user, ["admin"], []]]) + + if ql.startswith("show procedures"): + return Result(["name", "description", "signature"], [list(p) for p in PROCEDURES]) + + if ql.startswith("show functions"): + return Result(["name", "description", "signature"], []) + + if re.match(r"^show (range |lookup |fulltext |text |point |vector |all )?indexes", ql) \ + or ql.startswith("call db.indexes()"): + return Result(["id", "name", "state", "populationPercent", "type", "entityType", + "labelsOrTypes", "properties", "indexProvider", "owningConstraint"], []) + + if re.match(r"^show (all |unique |exist(ence)? |node key )?constraints", ql) \ + or ql.startswith("call db.constraints()"): + return Result(["id", "name", "type", "entityType", "labelsOrTypes", "properties", + "ownedIndex", "propertyType"], []) + + m = re.match(r"^show (?:(home|default) database|databases|database (`[^`]+`|[\w.\-]+))(?!\w)", + q, re.IGNORECASE) + if m: + return await self._show_databases(m.group(1), m.group(2)) + + m = re.match(r"^create (?:database|graph) (`[^`]+`|[\w.\-]+)( if not exists)?$", q, re.IGNORECASE) + if m: + return await self._create_graph(m.group(1).strip("`"), bool(m.group(2))) + + m = re.match(r"^drop (?:database|graph) (`[^`]+`|[\w.\-]+)( if exists)?", q, re.IGNORECASE) + if m: + return await self._drop_graph(m.group(1).strip("`"), bool(m.group(2))) + + if db == SYSTEM_DB: + return None + + # neo4j-browser's combined metadata queries (see dbMetaDuck.ts upstream). + if "call db.labels() yield label" in ql and "db.relationshiptypes()" in ql and "union all" in ql: + labels, types = await self._labels(db) + return Result(["result"], [[{"name": "labels", "data": labels}], + [{"name": "relationshipTypes", "data": types}], + [{"name": "propertyKeys", "data": await self._property_keys(db)}]]) + + if "return { name:'nodes', data:count(*) } as result" in ql: + nodes, rels = await self._counts(db) + return Result(["result"], [[{"name": "nodes", "data": nodes}], + [{"name": "relationships", "data": rels}]]) + + if re.match(r"^call db\.labels\(\)", ql): + labels, _ = await self._labels(db) + return Result(["label"], [[x] for x in labels]) + + if re.match(r"^call db\.relationshiptypes\(\)", ql): + _, types = await self._labels(db) + return Result(["relationshipType"], [[x] for x in types]) + + if re.match(r"^call db\.propertykeys\(\)", ql): + return Result(["propertyKey"], [[x] for x in await self._property_keys(db)]) + + if re.match(r"^call db\.schema\.visualization\(\)", ql): + return await self._schema_visualization(db) + + m = re.match(r"^call (db|dbms|apoc|gds)\.[\w.]+", ql) + if m: + raise BoltFailure("Neo.ClientError.Procedure.ProcedureNotFound", + f"There is no procedure with the name `{m.group(0)[5:]}` registered " + "for this database instance (peo4c).") + return None + + async def _show_databases(self, special, name): + graphs = await self.graphs() + default = self.config.default_graph + if default not in graphs and graphs: + default = graphs[0] + rows = [] + for g in [SYSTEM_DB] + graphs: + is_default = g == default + rows.append([g, "system" if g == SYSTEM_DB else "standard", [], "read-write", + self.config.advertised_bolt, "primary", True, "online", "online", "", + is_default, is_default, []]) + if special: + rows = [r for r in rows if r[10]] + elif name: + name = name.strip("`").lower() + rows = [r for r in rows if r[0].lower() == name] + return Result(["name", "type", "aliases", "access", "address", "role", "writer", + "requestedStatus", "currentStatus", "statusMessage", "default", "home", + "constituents"], rows) + + async def _create_graph(self, name, if_not_exists): + if not GRAPH_NAME.match(name): + raise BoltFailure("Neo.ClientError.Statement.ArgumentError", + f"Invalid graph name '{name}': AGE graph names must be 3-63 characters " + "of letters, digits, '_', '.' or '-', starting with a letter or '_'.") + if name in await self.graphs(): + if if_not_exists: + return Result([], query_type="s") + raise BoltFailure("Neo.ClientError.Database.ExistingDatabaseFound", + f"Failed to create the specified database '{name}': Database already exists.") + await self._sql("SELECT ag_catalog.create_graph(%s)", [name]) + return Result([], query_type="s") + + async def _drop_graph(self, name, if_exists): + graphs = {g.lower(): g for g in await self.graphs()} + if name.lower() not in graphs: + if if_exists: + return Result([], query_type="s") + raise BoltFailure("Neo.ClientError.Database.DatabaseNotFound", + f"Failed to delete the specified database '{name}': Database does not exist.") + await self._sql("SELECT ag_catalog.drop_graph(%s, true)", [graphs[name.lower()]]) + return Result([], query_type="s") + + async def _labels(self, graph): + rows = await self._sql( + "SELECT l.name, l.kind FROM ag_catalog.ag_label l " + "JOIN ag_catalog.ag_graph g ON l.graph = g.graphid " + "WHERE g.name = %s AND l.name <> ALL(%s) ORDER BY l.name", + [graph, list(SYSTEM_LABELS)]) + return [r[0] for r in rows if r[1] == "v"], [r[0] for r in rows if r[1] == "e"] + + async def _cypher_values(self, graph, query): + rows = await self._sql( + f"SELECT * FROM ag_catalog.cypher({sql.Literal(graph).as_string(self.conn)}, " + f"$peo4c$ {query} $peo4c$) AS (c0 ag_catalog.agtype)") + return [agtype.parse(r[0]) for r in rows] + + async def _property_keys(self, graph): + # Sample rather than scan everything: this runs every 20s from the browser. + keys = set() + for q in ("MATCH (n) WITH n LIMIT 1000 UNWIND keys(n) AS k RETURN DISTINCT k", + "MATCH ()-[r]->() WITH r LIMIT 1000 UNWIND keys(r) AS k RETURN DISTINCT k"): + keys.update(await self._cypher_values(graph, q)) + return sorted(k for k in keys if isinstance(k, str)) + + async def _counts(self, graph): + nodes = await self._sql(sql.SQL("SELECT count(*) FROM {}._ag_label_vertex").format(sql.Identifier(graph))) + rels = await self._sql(sql.SQL("SELECT count(*) FROM {}._ag_label_edge").format(sql.Identifier(graph))) + return nodes[0][0], rels[0][0] + + async def _schema_visualization(self, graph): + labels, _ = await self._labels(graph) + triples = await self._cypher_values( + graph, "MATCH (a)-[r]->(b) RETURN DISTINCT [label(a), label(r), label(b)]") + node_ids = {} + nodes = [] + for label in labels + [t[0] for t in triples] + [t[2] for t in triples]: + if label not in node_ids: + node_ids[label] = -(len(node_ids) + 1) + nodes.append(Node(node_ids[label], [label or "(no label)"], + {"name": label or "(no label)", "indexes": [], "constraints": []})) + rels = [Relationship(-(i + 1), node_ids[a], node_ids[b], t, {"name": t}) + for i, (a, t, b) in enumerate(triples)] + return Result(["nodes", "relationships"], [[nodes, rels]]) diff --git a/bridge/peo4c/bolt.py b/bridge/peo4c/bolt.py new file mode 100644 index 0000000..b5c7f11 --- /dev/null +++ b/bridge/peo4c/bolt.py @@ -0,0 +1,370 @@ +"""Bolt protocol server (raw TCP and WebSocket on the same port). + +Supports Bolt 4.4 and 5.0-5.4, which covers the neo4j-javascript-driver used +by neo4j-browser as well as current Python/Java drivers and cypher-shell. +""" + +import asyncio +import base64 +import hashlib +import itertools +import logging +import struct + +from . import packstream +from .backend import BoltFailure, Config, Session +from .packstream import Structure +from .web import discovery_document + +log = logging.getLogger("peo4c.bolt") + +MAGIC = b"\x60\x60\xB0\x17" +SUPPORTED = [(5, 4), (5, 3), (5, 2), (5, 1), (5, 0), (4, 4)] +SERVER_AGENT = "Neo4j/5.26.0" + +HELLO, GOODBYE, RESET, RUN, BEGIN, COMMIT, ROLLBACK = 0x01, 0x02, 0x0F, 0x10, 0x11, 0x12, 0x13 +DISCARD, PULL, TELEMETRY, ROUTE, LOGON, LOGOFF = 0x2F, 0x3F, 0x54, 0x66, 0x6A, 0x6B +SUCCESS, RECORD, IGNORED, FAILURE = 0x70, 0x71, 0x7E, 0x7F + +_connection_ids = itertools.count(1) + + +class ConnectionClosed(Exception): + pass + + +# -- transports ----------------------------------------------------------------- + + +class RawTransport: + def __init__(self, reader, writer, prefix=b""): + self.reader, self.writer, self.buffer = reader, writer, bytearray(prefix) + + async def read(self, n): + while len(self.buffer) < n: + chunk = await self.reader.read(65536) + if not chunk: + raise ConnectionClosed() + self.buffer += chunk + data = bytes(self.buffer[:n]) + del self.buffer[:n] + return data + + async def write(self, data): + self.writer.write(data) + await self.writer.drain() + + async def close(self): + self.writer.close() + + +class WebSocketTransport(RawTransport): + """Minimal RFC 6455 server: Bolt bytes are carried in binary frames.""" + + GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + + async def handshake(self): + request = bytearray(self.buffer) + self.buffer.clear() + while b"\r\n\r\n" not in request: + chunk = await self.reader.read(4096) + if not chunk or len(request) > 65536: + raise ConnectionClosed() + request += chunk + head, _, rest = bytes(request).partition(b"\r\n\r\n") + headers = {} + for line in head.decode("latin-1").split("\r\n")[1:]: + k, _, v = line.partition(":") + headers[k.strip().lower()] = v.strip() + method = head.split(b" ", 1)[0] + key = headers.get("sec-websocket-key") + if method != b"GET" or not key: + await self._plain_http(method, headers) + raise ConnectionClosed() + accept = base64.b64encode(hashlib.sha1((key + self.GUID).encode()).digest()).decode() + self.writer.write(("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n" + f"Connection: Upgrade\r\nSec-WebSocket-Accept: {accept}\r\n\r\n").encode()) + await self.writer.drain() + self.raw = bytearray(rest) + + async def _plain_http(self, method, headers): + """Answer non-WebSocket requests, e.g. neo4j-browser probing for discovery data.""" + cors = (b"Access-Control-Allow-Origin: *\r\nAccess-Control-Allow-Headers: *\r\n" + b"Access-Control-Allow-Methods: GET, HEAD, OPTIONS\r\nConnection: close\r\n") + if method == b"OPTIONS": + response = b"HTTP/1.1 204 No Content\r\n" + cors + b"\r\n" + elif method in (b"GET", b"HEAD"): + body = discovery_document(headers, self.writer.get_extra_info("sockname")[1]) + response = (b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n" + cors + + b"Content-Length: %d\r\n\r\n" % len(body) + (body if method == b"GET" else b"")) + else: + response = b"HTTP/1.1 405 Method Not Allowed\r\n" + cors + b"Content-Length: 0\r\n\r\n" + self.writer.write(response) + await self.writer.drain() + + async def _raw(self, n): + while len(self.raw) < n: + chunk = await self.reader.read(65536) + if not chunk: + raise ConnectionClosed() + self.raw += chunk + data = bytes(self.raw[:n]) + del self.raw[:n] + return data + + async def read(self, n): + while len(self.buffer) < n: + b0, b1 = await self._raw(2) + opcode, masked, length = b0 & 0x0F, b1 & 0x80, b1 & 0x7F + if length == 126: + length = struct.unpack(">H", await self._raw(2))[0] + elif length == 127: + length = struct.unpack(">Q", await self._raw(8))[0] + mask = await self._raw(4) if masked else None + payload = await self._raw(length) + if mask: + payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + if opcode == 0x8: + await self._frame(0x8, payload[:2]) + raise ConnectionClosed() + if opcode == 0x9: + await self._frame(0xA, payload) + elif opcode in (0x0, 0x1, 0x2): + self.buffer += payload + data = bytes(self.buffer[:n]) + del self.buffer[:n] + return data + + async def _frame(self, opcode, payload): + n = len(payload) + if n < 126: + header = struct.pack(">BB", 0x80 | opcode, n) + elif n < 0x10000: + header = struct.pack(">BBH", 0x80 | opcode, 126, n) + else: + header = struct.pack(">BBQ", 0x80 | opcode, 127, n) + self.writer.write(header + payload) + await self.writer.drain() + + async def write(self, data): + await self._frame(0x2, data) + + +# -- protocol ----------------------------------------------------------------------- + + +def negotiate(proposals: bytes): + for i in range(4): + _, span, minor, major = proposals[i * 4:i * 4 + 4] + for version in SUPPORTED: + if version[0] == major and minor - span <= version[1] <= minor: + return version + return None + + +class BoltConnection: + def __init__(self, transport, config: Config): + self.t = transport + self.config = config + self.id = f"bolt-{next(_connection_ids)}" + self.version = None + self.packer = None + self.session: Session | None = None + self.failed = False + self.routing = {} + self.results = {} # qid -> [Result, next row index] + self.last_qid = -1 + self.qids = itertools.count() + + async def serve(self): + proposals = await self.t.read(16) + self.version = negotiate(proposals) + if self.version is None: + log.info("%s: no supported Bolt version in %s", self.id, proposals.hex()) + await self.t.write(b"\x00\x00\x00\x00") + return + await self.t.write(bytes([0, 0, self.version[1], self.version[0]])) + self.packer = packstream.Packer(element_ids=self.version >= (5, 0)) + log.debug("%s: negotiated Bolt %d.%d", self.id, *self.version) + try: + while True: + message = await self.read_message() + if message is None: + continue # NOOP chunk + if not await self.handle(message): + break + finally: + if self.session: + await self.session.close() + + async def read_message(self): + data = bytearray() + while True: + size = struct.unpack(">H", await self.t.read(2))[0] + if size == 0: + if not data: + return None + return packstream.unpack(bytes(data)) + data += await self.t.read(size) + + async def send(self, tag, *fields): + data = self.packer.pack(Structure(tag, list(fields))) + out = bytearray() + for i in range(0, len(data), 0xFFFF): + chunk = data[i:i + 0xFFFF] + out += struct.pack(">H", len(chunk)) + chunk + out += b"\x00\x00" + await self.t.write(bytes(out)) + + async def success(self, metadata=None): + await self.send(SUCCESS, metadata or {}) + + async def failure(self, code, message): + log.info("%s: %s: %s", self.id, code, message) + self.failed = True + await self.send(FAILURE, {"code": code, "message": message}) + + async def handle(self, msg) -> bool: + tag, f = msg.tag, msg.fields + if tag == GOODBYE: + return False + if tag == RESET: + self.failed = False + self.results.clear() + if self.session: + await self.session.reset() + await self.success() + return True + if self.failed: + await self.send(IGNORED) + return True + try: + await self.dispatch(tag, f) + except BoltFailure as e: + self.results.clear() + await self.failure(e.code, e.message) + except Exception as e: # noqa: BLE001 - report anything to the client + log.exception("%s: internal error", self.id) + self.results.clear() + await self.failure("Neo.DatabaseError.General.UnknownError", f"peo4c internal error: {e}") + return True + + async def dispatch(self, tag, f): + if tag == HELLO: + extra = f[0] if f else {} + self.routing = extra.get("routing") or {} + if self.version < (5, 1): + await self.logon(extra) + await self.success({"server": SERVER_AGENT, "connection_id": self.id, "hints": {}}) + elif tag == LOGON: + await self.logon(f[0]) + await self.success() + elif tag == LOGOFF: + if self.session: + await self.session.close() + self.session = None + await self.success() + elif tag == TELEMETRY: + await self.success() + elif tag == ROUTE: + await self.route(f) + elif tag == RUN: + await self.run(f) + elif tag in (PULL, DISCARD): + await self.pull(f[0] if f else {}, discard=tag == DISCARD) + elif tag == BEGIN: + extra = f[0] if f else {} + await self.require_session().begin(extra.get("db")) + self.results.clear() + await self.success() + elif tag == COMMIT: + await self.require_session().commit() + await self.success({"bookmark": "peo4c:bookmark"}) + elif tag == ROLLBACK: + await self.require_session().rollback() + await self.success() + else: + raise BoltFailure("Neo.ClientError.Request.Invalid", f"Unsupported Bolt message 0x{tag:02X}") + + def require_session(self) -> Session: + if self.session is None: + raise BoltFailure("Neo.ClientError.Security.Unauthorized", "Not authenticated") + return self.session + + async def logon(self, auth): + if auth.get("scheme") != "basic": + raise BoltFailure("Neo.ClientError.Security.Unauthorized", + "peo4c requires basic authentication with a PostgreSQL user name and password.") + session = Session(self.config, auth.get("principal", "")) + await session.connect(auth.get("credentials", "")) + self.session = session + + async def route(self, f): + extra = f[2] if len(f) > 2 else None + db = extra.get("db") if isinstance(extra, dict) else extra + db = await self.require_session().resolve_db(db) + address = self.routing.get("address") or self.config.advertised_bolt + await self.success({"rt": {"ttl": 300, "db": db, "servers": [ + {"addresses": [address], "role": "WRITE"}, + {"addresses": [address], "role": "READ"}, + {"addresses": [address], "role": "ROUTE"}, + ]}}) + + async def run(self, f): + query, params = f[0], f[1] if len(f) > 1 else {} + extra = f[2] if len(f) > 2 else {} + session = self.require_session() + if not session.in_transaction: + self.results.clear() + result = await session.run(query, params, extra.get("db")) + qid = next(self.qids) + self.results[qid] = [result, 0] + self.last_qid = qid + metadata = {"fields": result.fields, "t_first": result.t_first} + if session.in_transaction: + metadata["qid"] = qid + await self.success(metadata) + + async def pull(self, extra, discard): + qid = extra.get("qid", -1) + qid = self.last_qid if qid == -1 else qid + if qid not in self.results: + raise BoltFailure("Neo.ClientError.Request.Invalid", "No result available to pull from") + result, index = self.results[qid] + n = extra.get("n", -1) + end = len(result.rows) if n == -1 or discard else min(index + n, len(result.rows)) + if not discard: + for row in result.rows[index:end]: + await self.send(RECORD, row) + self.results[qid][1] = end + if end < len(result.rows): + await self.success({"has_more": True}) + return + del self.results[qid] + metadata = {"type": result.query_type, "t_last": 0, "db": result.db} + if not self.session.in_transaction: + metadata["bookmark"] = "peo4c:bookmark" + await self.success(metadata) + + +async def handle_client(reader, writer, config: Config): + peer = writer.get_extra_info("peername") + try: + first = await reader.readexactly(4) + log.debug("connection from %s starts with %r", peer, first) + if first == MAGIC: + transport = RawTransport(reader, writer) + elif first in (b"GET ", b"HEAD", b"OPTI", b"POST"): + transport = WebSocketTransport(reader, writer, first) + await transport.handshake() + if await transport.read(4) != MAGIC: + return + else: + return + await BoltConnection(transport, config).serve() + except (ConnectionClosed, asyncio.IncompleteReadError, ConnectionResetError, BrokenPipeError): + pass + except Exception: # noqa: BLE001 + log.exception("connection from %s failed", peer) + finally: + writer.close() diff --git a/bridge/peo4c/cypher.py b/bridge/peo4c/cypher.py new file mode 100644 index 0000000..fa75759 --- /dev/null +++ b/bridge/peo4c/cypher.py @@ -0,0 +1,236 @@ +"""Light-weight Cypher text analysis needed to run queries through AGE. + +AGE's ``cypher()`` SQL function needs the caller to declare the result columns +up front, does not accept Neo4j-style parameter maps reliably, and lacks a few +constructs neo4j-browser relies on. This module works on the query text only: +it masks out strings and comments, finds the final top-level RETURN clause, +and inlines parameters as Cypher literals. +""" + +import math +import re +from dataclasses import dataclass + + +class CypherError(Exception): + def __init__(self, code, message): + super().__init__(message) + self.code = code + self.message = message + + +@dataclass +class Scan: + text: str + masked: str # strings/backticks replaced by 'x', comments by spaces + depth: list # bracket nesting depth for every character + + +def scan(text: str) -> Scan: + masked = list(text) + depth = [0] * len(text) + level = 0 + i, n = 0, len(text) + while i < n: + c = text[i] + if c in "'\"`": + j = i + 1 + while j < n and text[j] != c: + j += 2 if text[j] == "\\" and c != "`" else 1 + for k in range(i, min(j + 1, n)): + masked[k] = "x" + depth[k] = level + i = j + 1 + continue + if text.startswith("//", i): + j = text.find("\n", i) + j = n if j == -1 else j + for k in range(i, j): + masked[k] = " " + depth[k] = level + i = j + continue + if text.startswith("/*", i): + j = text.find("*/", i + 2) + j = n if j == -1 else j + 2 + for k in range(i, j): + masked[k] = " " + depth[k] = level + i = j + continue + if c in "([{": + depth[i] = level + level += 1 + elif c in ")]}": + level = max(level - 1, 0) + depth[i] = level + else: + depth[i] = level + i += 1 + return Scan(text, "".join(masked), depth) + + +def _top_level(s: Scan, pattern, start=0, end=None): + end = len(s.text) if end is None else end + for m in re.finditer(pattern, s.masked[start:end], re.IGNORECASE): + if s.depth[start + m.start()] == 0: + yield start + m.start(), start + m.end() + + +def strip_statement(text: str) -> str: + text = text.strip() + while text.endswith(";"): + text = text[:-1].rstrip() + return text + + +def return_columns(text: str) -> list: + """Column names of the final top-level RETURN clause ([] if there is none).""" + s = scan(text) + returns = list(_top_level(s, r"(? str: + """Replace ``$name`` references with literal Cypher values.""" + if "$" not in text: + return text + s = scan(text) + pieces, last, missing = [], 0, [] + for m in _PARAM.finditer(s.masked): + name = m.group(1) + if name.startswith("`"): + name = name[1:-1].replace("``", "`") + if name not in params: + missing.append(name) + continue + pieces.append(text[last:m.start()]) + pieces.append(literal(params[name])) + last = m.end() + if missing: + raise CypherError("Neo.ClientError.Statement.ParameterMissing", + "Expected parameter(s): " + ", ".join(dict.fromkeys(missing))) + pieces.append(text[last:]) + return "".join(pieces) + + +def _escape_string(v: str) -> str: + out = [] + for ch in v: + if ch == "\\": + out.append("\\\\") + elif ch == "'": + out.append("\\'") + elif ch == "\n": + out.append("\\n") + elif ch == "\r": + out.append("\\r") + elif ch == "\t": + out.append("\\t") + elif ord(ch) < 0x20: + out.append(f"\\u{ord(ch):04x}") + else: + out.append(ch) + return "'" + "".join(out) + "'" + + +def literal(v) -> str: + if v is None: + return "null" + if v is True: + return "true" + if v is False: + return "false" + if isinstance(v, int): + return str(v) + if isinstance(v, float): + if math.isnan(v): + return "'NaN'::float" + if math.isinf(v): + return "'Infinity'::float" if v > 0 else "'-Infinity'::float" + return repr(v) + if isinstance(v, str): + return _escape_string(v) + if isinstance(v, (list, tuple)): + return "[" + ", ".join(literal(x) for x in v) + "]" + if isinstance(v, dict): + return "{" + ", ".join( + "`" + str(k).replace("`", "``") + "`: " + literal(x) for k, x in v.items()) + "}" + raise CypherError("Neo.ClientError.Statement.TypeError", + f"Parameters of type {type(v).__name__} are not supported by peo4c") + + +# neo4j-browser's "expand node" query uses a pattern comprehension, which AGE +# does not support. Rewrite it into an equivalent OPTIONAL MATCH + count. +_NEIGHBOUR_COUNT = re.compile( + r"WITH\s+(\w+)\s*,\s*size\(\s*\[\s*\(\s*\1\s*\)\s*--\s*\(\s*\)\s*\|\s*1\s*\]\s*\)\s+AS\s+(\w+)", + re.IGNORECASE) + + +def rewrite_for_age(text: str) -> str: + return _NEIGHBOUR_COUNT.sub( + lambda m: f"OPTIONAL MATCH ({m.group(1)})-[peo4c_r]-() " + f"WITH {m.group(1)}, count(peo4c_r) AS {m.group(2)}", + text) + + +# AGE fails with "not a common type" on `id(x) IN [a, b, ...]` when the list has +# more than one element, which neo4j-browser uses to fetch relationships between +# displayed nodes. Expand such lists of integer literals into an OR chain. +_ID_IN_LIST = re.compile(r"(? str: + s = scan(text) + pieces, last = [], 0 + for m in _ID_IN_LIST.finditer(s.masked): + values = [v.strip() for v in m.group(2).split(",") if v.strip()] + if len(values) == 1: + continue + expr = text[m.start(1):m.end(1)] + pieces.append(text[last:m.start()]) + pieces.append("(" + " OR ".join(f"{expr} = {v}" for v in values) + ")" if values else "false") + last = m.end() + pieces.append(text[last:]) + return "".join(pieces) + + +def first_keyword(text: str) -> str: + m = re.match(r"\s*(\w+)", scan(text).masked) + return m.group(1).upper() if m else "" diff --git a/bridge/peo4c/packstream.py b/bridge/peo4c/packstream.py new file mode 100644 index 0000000..f4b32cd --- /dev/null +++ b/bridge/peo4c/packstream.py @@ -0,0 +1,241 @@ +"""PackStream v1 serialisation, the binary value format used by Bolt.""" + +import struct +from dataclasses import dataclass, field + + +@dataclass +class Structure: + tag: int + fields: list = field(default_factory=list) + + +@dataclass +class Node: + id: int + labels: list + properties: dict + + +@dataclass +class Relationship: + id: int + start: int + end: int + type: str + properties: dict + + +@dataclass +class Path: + nodes: list # alternating sequence is [n0, r0, n1, r1, ..., nk] + relationships: list + + +class PackStreamError(Exception): + pass + + +class Packer: + """Encodes Python values. ``element_ids`` selects the Bolt 5 graph struct layout.""" + + def __init__(self, element_ids: bool = True): + self.element_ids = element_ids + self.out = bytearray() + + def pack(self, value) -> bytes: + self.out = bytearray() + self._pack(value) + return bytes(self.out) + + def _pack(self, v): + out = self.out + if v is None: + out.append(0xC0) + elif v is True: + out.append(0xC3) + elif v is False: + out.append(0xC2) + elif isinstance(v, int): + self._pack_int(v) + elif isinstance(v, float): + out.append(0xC1) + out += struct.pack(">d", v) + elif isinstance(v, str): + data = v.encode("utf-8") + self._header(len(data), 0x80, 0xD0, 0xD1, 0xD2) + out += data + elif isinstance(v, (bytes, bytearray)): + n = len(v) + if n < 0x100: + out += bytes([0xCC, n]) + elif n < 0x10000: + out.append(0xCD) + out += struct.pack(">H", n) + else: + out.append(0xCE) + out += struct.pack(">I", n) + out += v + elif isinstance(v, (list, tuple)): + self._header(len(v), 0x90, 0xD4, 0xD5, 0xD6) + for item in v: + self._pack(item) + elif isinstance(v, dict): + self._header(len(v), 0xA0, 0xD8, 0xD9, 0xDA) + for k, item in v.items(): + self._pack(str(k)) + self._pack(item) + elif isinstance(v, Structure): + if len(v.fields) > 15: + raise PackStreamError("structure too large") + out.append(0xB0 + len(v.fields)) + out.append(v.tag) + for item in v.fields: + self._pack(item) + elif isinstance(v, Node): + self._pack(self._node(v)) + elif isinstance(v, Relationship): + fields = [v.id, v.start, v.end, v.type, v.properties] + if self.element_ids: + fields += [str(v.id), str(v.start), str(v.end)] + self._pack(Structure(0x52, fields)) + elif isinstance(v, Path): + self._pack(self._path(v)) + else: + raise PackStreamError(f"cannot pack value of type {type(v).__name__}") + + def _node(self, n: Node) -> Structure: + fields = [n.id, n.labels, n.properties] + if self.element_ids: + fields.append(str(n.id)) + return Structure(0x4E, fields) + + def _path(self, p: Path) -> Structure: + nodes, node_index = [], {} + for n in p.nodes: + if n.id not in node_index: + node_index[n.id] = len(nodes) + nodes.append(n) + rels, rel_index = [], {} + for r in p.relationships: + if r.id not in rel_index: + rel_index[r.id] = len(rels) + fields = [r.id, r.type, r.properties] + if self.element_ids: + fields.append(str(r.id)) + rels.append(Structure(0x72, fields)) + indices = [] + prev = p.nodes[0] + for r, n in zip(p.relationships, p.nodes[1:]): + idx = rel_index[r.id] + 1 + indices.append(idx if r.start == prev.id else -idx) + indices.append(node_index[n.id]) + prev = n + return Structure(0x50, [[self._node(n) for n in nodes], rels, indices]) + + def _header(self, n, tiny, s8, s16, s32): + out = self.out + if n < 0x10: + out.append(tiny + n) + elif n < 0x100: + out += bytes([s8, n]) + elif n < 0x10000: + out.append(s16) + out += struct.pack(">H", n) + else: + out.append(s32) + out += struct.pack(">I", n) + + def _pack_int(self, v): + out = self.out + if -0x10 <= v < 0x80: + out += struct.pack(">b", v) + elif -0x80 <= v < 0x80: + out.append(0xC8) + out += struct.pack(">b", v) + elif -0x8000 <= v < 0x8000: + out.append(0xC9) + out += struct.pack(">h", v) + elif -0x80000000 <= v < 0x80000000: + out.append(0xCA) + out += struct.pack(">i", v) + elif -0x8000000000000000 <= v < 0x8000000000000000: + out.append(0xCB) + out += struct.pack(">q", v) + else: + raise PackStreamError("integer out of 64-bit range") + + +class Unpacker: + def __init__(self, data: bytes): + self.data = data + self.pos = 0 + + def _take(self, n): + if self.pos + n > len(self.data): + raise PackStreamError("unexpected end of data") + chunk = self.data[self.pos:self.pos + n] + self.pos += n + return chunk + + def _u(self, fmt, n): + return struct.unpack(fmt, self._take(n))[0] + + def unpack(self): + marker = self._take(1)[0] + if marker < 0x80: + return marker + if marker >= 0xF0: + return marker - 0x100 + hi = marker & 0xF0 + if hi == 0x80: + return self._take(marker & 0x0F).decode("utf-8") + if hi == 0x90: + return [self.unpack() for _ in range(marker & 0x0F)] + if hi == 0xA0: + return self._map(marker & 0x0F) + if hi == 0xB0: + tag = self._take(1)[0] + return Structure(tag, [self.unpack() for _ in range(marker & 0x0F)]) + if marker == 0xC0: + return None + if marker == 0xC1: + return self._u(">d", 8) + if marker == 0xC2: + return False + if marker == 0xC3: + return True + if marker == 0xC8: + return self._u(">b", 1) + if marker == 0xC9: + return self._u(">h", 2) + if marker == 0xCA: + return self._u(">i", 4) + if marker == 0xCB: + return self._u(">q", 8) + if marker in (0xCC, 0xCD, 0xCE): + n = {0xCC: lambda: self._u(">B", 1), 0xCD: lambda: self._u(">H", 2), + 0xCE: lambda: self._u(">I", 4)}[marker]() + return bytes(self._take(n)) + if marker in (0xD0, 0xD1, 0xD2): + return self._take(self._size(marker - 0xD0)).decode("utf-8") + if marker in (0xD4, 0xD5, 0xD6): + return [self.unpack() for _ in range(self._size(marker - 0xD4))] + if marker in (0xD8, 0xD9, 0xDA): + return self._map(self._size(marker - 0xD8)) + raise PackStreamError(f"unknown marker 0x{marker:02X}") + + def _size(self, width_code): + return [lambda: self._u(">B", 1), lambda: self._u(">H", 2), + lambda: self._u(">I", 4)][width_code]() + + def _map(self, n): + result = {} + for _ in range(n): + key = self.unpack() + result[key] = self.unpack() + return result + + +def unpack(data: bytes): + return Unpacker(data).unpack() diff --git a/bridge/peo4c/web.py b/bridge/peo4c/web.py new file mode 100644 index 0000000..c39e366 --- /dev/null +++ b/bridge/peo4c/web.py @@ -0,0 +1,84 @@ +"""Tiny HTTP server for the neo4j-browser static files and Neo4j discovery.""" + +import asyncio +import json +import logging +import mimetypes +import os +from pathlib import Path +from urllib.parse import unquote, urlsplit + +from .backend import NEO4J_VERSION + +log = logging.getLogger("peo4c.http") + + +class HttpServer: + def __init__(self, browser_dir: str, bolt_port: int): + self.root = Path(browser_dir).resolve() + self.bolt_port = bolt_port + + async def handle(self, reader, writer): + try: + request = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=30) + lines = request.decode("latin-1").split("\r\n") + method, target, _ = (lines[0].split(" ") + ["", "", ""])[:3] + headers = {} + for line in lines[1:]: + k, _, v = line.partition(":") + headers[k.strip().lower()] = v.strip() + status, extra, body = self.respond(method, target, headers) + head = [f"HTTP/1.1 {status}", f"Content-Length: {len(body)}", "Connection: close", + "X-Content-Type-Options: nosniff"] + head += [f"{k}: {v}" for k, v in extra.items()] + writer.write(("\r\n".join(head) + "\r\n\r\n").encode("latin-1")) + if method != "HEAD": + writer.write(body) + await writer.drain() + except (asyncio.IncompleteReadError, asyncio.LimitOverrunError, asyncio.TimeoutError, + ConnectionResetError, BrokenPipeError): + pass + except Exception: # noqa: BLE001 + log.exception("HTTP request failed") + finally: + writer.close() + + def respond(self, method, target, headers): + if method not in ("GET", "HEAD"): + return "405 Method Not Allowed", {"Allow": "GET, HEAD"}, b"" + path = unquote(urlsplit(target).path) + + if path in ("/", ""): + if "text/html" in headers.get("accept", ""): + return "302 Found", {"Location": "/browser/"}, b"" + return "200 OK", {"Content-Type": "application/json"}, discovery_document(headers, self.bolt_port) + + if path == "/browser": + return "301 Moved Permanently", {"Location": "/browser/"}, b"" + if not path.startswith("/browser/"): + return "404 Not Found", {"Content-Type": "text/plain"}, b"Not found\n" + + file = (self.root / path[len("/browser/"):]).resolve() + if file.is_dir(): + file = file / "index.html" + if not file.is_relative_to(self.root) or not file.is_file(): + return "404 Not Found", {"Content-Type": "text/plain"}, b"Not found\n" + ctype = mimetypes.guess_type(file.name)[0] or "application/octet-stream" + cache = "no-cache" if file.name.endswith(".html") else "public, max-age=86400" + return "200 OK", {"Content-Type": ctype, "Cache-Control": cache}, file.read_bytes() + + +def discovery_document(headers, bolt_port) -> bytes: + """The JSON Neo4j serves at ``/`` so clients can find the Bolt endpoint.""" + host = headers.get("host") or "localhost" + if host.startswith("["): # IPv6 literal, e.g. [::1]:7474 + host = host[:host.index("]") + 1] + else: + host = host.split(":", 1)[0] + port = os.environ.get("PEO4C_ADVERTISED_BOLT_PORT", str(bolt_port)) + return json.dumps({ + "bolt_routing": f"neo4j://{host}:{port}", + "bolt_direct": f"bolt://{host}:{port}", + "neo4j_version": NEO4J_VERSION, + "neo4j_edition": "community", + }).encode() diff --git a/docker/initdb/10-peo4c.sh b/docker/initdb/10-peo4c.sh new file mode 100755 index 0000000..b485dcc --- /dev/null +++ b/docker/initdb/10-peo4c.sh @@ -0,0 +1,18 @@ +#!/usr/bin/env bash +# Prepares a freshly initialised cluster for peo4c. +set -euo pipefail + +# The bridge connects over 127.0.0.1 with the credentials users log in with, so +# loopback TCP connections must check passwords. The postgres image trusts them +# by default; Unix-socket connections (psql inside the container) stay trusted. +sed -i -E 's/^(host[[:space:]]+[^[:space:]]+[[:space:]]+[^[:space:]]+[[:space:]]+(127\.0\.0\.1\/32|::1\/128)[[:space:]]+)trust/\1scram-sha-256/' \ + "$PGDATA/pg_hba.conf" + +# Create the default AGE graph that neo4j-browser opens on first connect. +psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "${POSTGRES_DB:-$POSTGRES_USER}" \ + -v graph="${PEO4C_DEFAULT_GRAPH:-neo4j}" <<'EOSQL' +LOAD 'age'; +SET search_path = ag_catalog, "$user", public; +SELECT ag_catalog.create_graph(:'graph') +WHERE NOT EXISTS (SELECT 1 FROM ag_catalog.ag_graph WHERE name = :'graph'); +EOSQL diff --git a/docker/peo4c-entrypoint.sh b/docker/peo4c-entrypoint.sh new file mode 100755 index 0000000..9f93e74 --- /dev/null +++ b/docker/peo4c-entrypoint.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Runs PostgreSQL (through the stock postgres entrypoint) and the peo4c bridge +# side by side, and stops the container when either of them exits. +set -euo pipefail + +if [ "${1:-}" != "postgres" ]; then + exec docker-entrypoint.sh "$@" +fi + +export PEO4C_PG_DATABASE="${PEO4C_PG_DATABASE:-${POSTGRES_DB:-${POSTGRES_USER:-postgres}}}" + +docker-entrypoint.sh "$@" & +pg_pid=$! + +cd /opt/peo4c/bridge +gosu postgres python3 -m peo4c & +bridge_pid=$! + +shutdown() { + kill -TERM "$bridge_pid" 2>/dev/null || true + kill -TERM "$pg_pid" 2>/dev/null || true +} +trap shutdown TERM INT + +# Wait until one of the two processes exits, then stop the other one too. +set +e +wait -n "$pg_pid" "$bridge_pid" +status=$? +shutdown +wait "$pg_pid" 2>/dev/null +exit "$status" diff --git a/tests/smoke_test.py b/tests/smoke_test.py new file mode 100644 index 0000000..4e22619 --- /dev/null +++ b/tests/smoke_test.py @@ -0,0 +1,121 @@ +"""End-to-end smoke test for a running peo4c bridge, using the official Neo4j driver. + +Usage: python tests/smoke_test.py [bolt-uri] [user] [password] +Requires: pip install neo4j +""" + +import sys + +import neo4j +from neo4j.graph import Node, Path, Relationship + +uri = sys.argv[1] if len(sys.argv) > 1 else "neo4j://localhost:7687" +user = sys.argv[2] if len(sys.argv) > 2 else "postgres" +password = sys.argv[3] if len(sys.argv) > 3 else "postgres" +GRAPH = "peo4c_smoke" + + +def check(label, condition): + print(("ok " if condition else "FAIL ") + label) + if not condition: + check.failed = True + + +check.failed = False + +with neo4j.GraphDatabase.driver(uri, auth=(user, password)) as driver: + driver.verify_connectivity() + + with driver.session(database="system") as s: + s.run(f"DROP DATABASE {GRAPH} IF EXISTS").consume() + s.run(f"CREATE DATABASE {GRAPH}").consume() + names = [r["name"] for r in s.run("SHOW DATABASES")] + check("SHOW DATABASES lists system and new graph", "system" in names and GRAPH in names) + comp = s.run("CALL dbms.components() YIELD name, versions, edition").single() + check("dbms.components", comp["name"] == "Neo4j Kernel") + + with driver.session(database=GRAPH) as s: + s.run("CREATE (a:Person {name: $a, age: 30})-[:KNOWS {since: 2020}]->(b:Person {name: $b})", + a="Ann", b="Bob's").consume() + s.run("CREATE (:City {name: 'Utrecht', pop: 1.5})").consume() + + rec = s.run("MATCH (n:Person) WHERE n.name = $name RETURN n, n.age AS age", name="Ann").single() + check("node returned as Node", isinstance(rec["n"], Node) and rec["n"]["name"] == "Ann") + check("labels mapped", set(rec["n"].labels) == {"Person"}) + check("alias column", rec["age"] == 30) + + rec = s.run("MATCH p=(a)-[r:KNOWS]->(b) RETURN p, r, b.name").single() + check("path returned as Path", isinstance(rec["p"], Path) and len(rec["p"]) == 1) + check("relationship returned", isinstance(rec["r"], Relationship) and rec["r"].type == "KNOWS") + check("unaliased column name", rec.keys() == ["p", "r", "b.name"] and rec["b.name"] == "Bob's") + + ids = [r["id"] for r in s.run("MATCH (n) RETURN id(n) AS id")] + rels = list(s.run("MATCH (a)-[r]->(b) WHERE id(a) IN $existingNodeIds AND id(b) IN $newNodeIds " + "RETURN r;", existingNodeIds=ids, newNodeIds=ids)) + check("browser relationship autocomplete query", len(rels) == 1) + + ann = s.run("MATCH (n {name: 'Ann'}) RETURN id(n) AS id").single()["id"] + expand = list(s.run(f"""MATCH (a) WHERE id(a) = {ann} +WITH a, size([(a)--() | 1]) AS allNeighboursCount +MATCH path = (a)--(o) WHERE NOT id(o) IN [] +RETURN path, allNeighboursCount +ORDER BY id(o) +LIMIT 100""")) + check("browser expand-node query", len(expand) == 1 and expand[0]["allNeighboursCount"] == 1) + + meta = list(s.run(""" +CALL db.labels() YIELD label +RETURN {name:'labels', data:COLLECT(label)[..1000]} AS result +UNION ALL +CALL db.relationshipTypes() YIELD relationshipType +RETURN {name:'relationshipTypes', data:COLLECT(relationshipType)[..1000]} AS result +UNION ALL +CALL db.propertyKeys() YIELD propertyKey +RETURN {name:'propertyKeys', data:COLLECT(propertyKey)[..1000]} AS result +""")) + check("browser metadata query", meta[0]["result"]["data"] == ["City", "Person"] + and meta[1]["result"]["data"] == ["KNOWS"] and "name" in meta[2]["result"]["data"]) + + counts = list(s.run(""" +MATCH () RETURN { name:'nodes', data:count(*) } AS result +UNION ALL +MATCH ()-[]->() RETURN { name:'relationships', data: count(*)} AS result +""")) + check("browser count query", [c["result"]["data"] for c in counts] == [3, 1]) + + schema = s.run("CALL db.schema.visualization()").single() + check("schema visualization", len(schema["nodes"]) == 2 and len(schema["relationships"]) == 1) + + rows = list(s.run("UNWIND range(1, 2500) AS i RETURN i")) + check("large result over several PULLs", len(rows) == 2500 and rows[-1]["i"] == 2500) + + check("string escaping roundtrip", + s.run("RETURN $v AS v", v="a'b\\c\n\"d\" $x").single()["v"] == "a'b\\c\n\"d\" $x") + check("map/list params", + s.run("RETURN $m AS m", m={"k": [1, 2.5, None, True]}).single()["m"] == {"k": [1, 2.5, None, True]}) + + try: + s.run("MATCH (n RETURN n").consume() + check("syntax error reported", False) + except neo4j.exceptions.CypherSyntaxError: + check("syntax error reported", True) + + # The session must still be usable after a failure. + check("recovers after error", s.run("RETURN 1 AS x").single()["x"] == 1) + + with driver.session(database=GRAPH) as s: + with s.begin_transaction() as tx: + tx.run("CREATE (:Temp {n: 1})").consume() + tx.rollback() + check("explicit rollback", s.run("MATCH (n:Temp) RETURN count(n) AS c").single()["c"] == 0) + + def work(tx): + tx.run("CREATE (:Temp {n: 2})").consume() + return tx.run("MATCH (n:Temp) RETURN count(n) AS c").single()["c"] + + check("managed write transaction", s.execute_write(work) == 1) + + with driver.session(database="system") as s: + s.run(f"DROP DATABASE {GRAPH}").consume() + +sys.exit(1 if check.failed else 0)