More adversarial-testing fixes: race conditions, config schema validation, symlink cycles

cli/Cargo.toml

@@ -28,6 +28,9 @@ tar = "0.4"
 flate2 = "1"
 ignore = "0.4"
 
+[target.'cfg(unix)'.dependencies]
+libc = "0.2"
+
 [profile.release]
 opt-level = 2
 

docs/search-and-browse.md

@@ -135,8 +135,9 @@ Grep JSON is smaller:
 ```
 
 `via` can identify how grep found the match, for example pushdown, BM25, linear
-scan, or a notice. A `bm25` match is keyword-based, not an exact-literal or
-regex match — expect token-level matching rather than character-for-character
+scan, or a notice. A `bm25` match is keyword-based, not an exact-literal match;
+grep across all connectors today matches literal substrings/keywords, not
+regex patterns — expect token-level matching rather than character-for-character
 matches on database-backed sources.
 
 ## Reopen File-Like Hits

server/python/src/mfs_server/api/app.py

@@ -63,6 +63,7 @@
     ],
     "sync_already_running": ["mfs job list", "mfs job cancel JOB_ID"],
     "connector_removing": ["wait for removal to finish, then retry"],
+    "connector_already_registered": ["mfs connector list", "mfs connector inspect URI"],
     "remove_requires_connector_root": [
         "pass the registered connector root from `mfs connector list` or `mfs connector inspect`"
     ],
@@ -354,7 +355,12 @@ async def add(body: AddRequest) -> AddResponse:
             )
         except ValueError as e:
             code = str(e)
-            status = 409 if code in ("sync_already_running", "connector_removing") else 400
+            status = (
+                409
+                if code
+                in ("sync_already_running", "connector_removing", "connector_already_registered")
+                else 400
+            )
             raise HTTPException(status, code)  # -> error envelope
         return AddResponse(job_id=job_id)
 

server/python/src/mfs_server/api/models.py

@@ -166,6 +166,13 @@ class LsEntry(BaseModel):
         description="eligible for indexing (not binary, not opted out) — not a promise it is "
         "searchable; whether chunks were actually embedded is reported by search_status",
     )
+    real_id: Optional[str] = Field(
+        None,
+        description="symlink-resolved identity for a `dir` entry (e.g. 'dev:ino'), stable "
+        "across different path spellings of the same real directory. null for `file` "
+        "entries and for connector types that don't supply one. A recursive tree walk "
+        "should treat a repeated real_id as a symlink cycle and stop recursing into it.",
+    )
 
 
 class LsResponse(BaseModel):

server/python/src/mfs_server/connectors/base.py

@@ -14,6 +14,8 @@
 from dataclasses import dataclass, field
 from typing import Any, Literal, Optional, Protocol
 
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
 # --- object_kind (framework-fixed) ---
 ObjectKind = Literal[
     "document",
@@ -441,13 +443,47 @@ def was_partial(self, path: str) -> bool:
         return path in self._partial
 
 
+class ConnectorConfigSchema(BaseModel):
+    """Base for a connector's `CONFIG_SCHEMA`: validated at add/update/probe
+    time, before any connection is attempted (ConnectorFactory.validate_config).
+    `extra="forbid"` catches a typo'd/unknown field instead of silently
+    ignoring it; per-field types catch a wrong-typed value (e.g. a quoted
+    number) with a clean, named error instead of a raw exception surfacing
+    deep in the plugin's own connect()/read path.
+
+    `credential_ref`/`_credential_ref` (legacy alias) and `objects`
+    ([[objects]] per-object overrides) are cross-cutting fields every
+    connector config may carry regardless of type — declared once here so
+    connector-specific schemas only need their own real fields."""
+
+    model_config = ConfigDict(extra="forbid")
+    credential_ref: Optional[str] = None
+    objects: list = Field(default_factory=list)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _normalize_legacy_credential_ref(cls, data: Any) -> Any:
+        # A leading-underscore key is a private-attribute name to pydantic, not
+        # a field -- it would hit extra="forbid" instead of being recognized as
+        # the same thing PluginBuilder.build() already treats as an alias for
+        # credential_ref. Fold it in before field validation runs.
+        if isinstance(data, dict) and "_credential_ref" in data:
+            data = dict(data)
+            data.setdefault("credential_ref", data.pop("_credential_ref"))
+        return data
+
+
 class ConnectorPlugin(ABC):
     # --- metadata (class attrs) ---
     NAME: str = ""
     URI_SCHEME: str = ""
     DISPLAY_NAME: str = ""
     PROMPT: str = ""
     CAPABILITIES: Capabilities = Capabilities()
+    # Optional[type[ConnectorConfigSchema]]: when set, ConnectorFactory
+    # validates a caller-supplied config against it at add/update/probe time.
+    # Connectors without one (SaaS connectors with no local top-level config
+    # shape yet) are unaffected -- this is opt-in per connector, not universal.
     CONFIG_SCHEMA: Optional[type] = None
 
     def __init__(self, config: Any, credential: Any, *, ctx: ConnectorContext):

server/python/src/mfs_server/connectors/file/plugin.py

@@ -428,7 +428,20 @@ async def list(self, path: str) -> list[Entry]:
             if spec.match_file(test):
                 continue
             if child.is_dir():
-                entries.append(Entry(name=child.name, type="dir"))
+                # (st_dev, st_ino) survives symlink resolution -- two different path
+                # spellings of the same real directory (e.g. a symlink cycle) land on
+                # the identical id, unlike the path string itself, which keeps
+                # growing forever around a cycle. Lets a directory-tree walker (CLI
+                # `tree`) detect "I've been at this real directory before" and stop
+                # recursing instead of following a symlink cycle indefinitely.
+                st = child.stat()
+                entries.append(
+                    Entry(
+                        name=child.name,
+                        type="dir",
+                        extra={"real_id": f"{st.st_dev}:{st.st_ino}"},
+                    )
+                )
             else:
                 stt = child.stat()
                 entries.append(

server/python/src/mfs_server/connectors/mongo/plugin.py

@@ -13,6 +13,7 @@
 
 from ..base import (
     Capabilities,
+    ConnectorConfigSchema,
     ConnectorPlugin,
     Entry,
     HealthStatus,
@@ -24,6 +25,13 @@
 )
 
 
+class MongoConfig(ConnectorConfigSchema):
+    uri: Optional[str] = None
+    database: Optional[str] = None
+    max_read_docs: int = 100000
+    cursor_field: Optional[str] = None
+
+
 class MongoPlugin(ConnectorPlugin):
     NAME = "mongo"
     URI_SCHEME = "mongo"
@@ -37,6 +45,7 @@ class MongoPlugin(ConnectorPlugin):
         delete_detection="full_scan",
         paged_cat=True,
     )
+    CONFIG_SCHEMA = MongoConfig
 
     def __init__(self, config, credential, *, ctx):
         super().__init__(config, credential, ctx=ctx)

server/python/src/mfs_server/connectors/mysql/plugin.py

@@ -13,6 +13,7 @@
 
 from ..base import (
     Capabilities,
+    ConnectorConfigSchema,
     ConnectorPlugin,
     Entry,
     GrepMatch,
@@ -27,6 +28,16 @@
 )
 
 
+class MySQLConfig(ConnectorConfigSchema):
+    host: str = "127.0.0.1"
+    port: int = 3306
+    user: str = "root"
+    password: Optional[str] = None
+    database: Optional[str] = None
+    max_read_rows: int = 100000
+    cursor_column: Optional[str] = None
+
+
 class MySQLPlugin(ConnectorPlugin):
     NAME = "mysql"
     URI_SCHEME = "mysql"
@@ -42,6 +53,7 @@ class MySQLPlugin(ConnectorPlugin):
         search_pushdown=False,
         paged_cat=True,
     )
+    CONFIG_SCHEMA = MySQLConfig
 
     def __init__(self, config, credential, *, ctx):
         super().__init__(config, credential, ctx=ctx)

server/python/src/mfs_server/connectors/postgres/plugin.py

@@ -14,6 +14,7 @@
 
 from ..base import (
     Capabilities,
+    ConnectorConfigSchema,
     ConnectorPlugin,
     Entry,
     GrepMatch,
@@ -28,6 +29,13 @@
 )
 
 
+class PostgresConfig(ConnectorConfigSchema):
+    dsn: Optional[str] = None
+    schemas: list[str] = ["public"]
+    max_read_rows: int = 100000
+    cursor_column: Optional[str] = None
+
+
 class PostgresPlugin(ConnectorPlugin):
     NAME = "postgres"
     URI_SCHEME = "postgres"
@@ -43,6 +51,7 @@ class PostgresPlugin(ConnectorPlugin):
         search_pushdown=False,
         paged_cat=True,
     )
+    CONFIG_SCHEMA = PostgresConfig
 
     def __init__(self, config, credential, *, ctx):
         super().__init__(config, credential, ctx=ctx)

server/python/src/mfs_server/connectors/s3/plugin.py

@@ -19,6 +19,7 @@
 
 from ..base import (
     Capabilities,
+    ConnectorConfigSchema,
     ConnectorPlugin,
     Entry,
     HealthStatus,
@@ -31,6 +32,15 @@
 from ..file.plugin import CODE_EXT, DOC_EXT, IMAGE_EXT, TEXTBLOB_EXT
 
 
+class S3Config(ConnectorConfigSchema):
+    bucket: Optional[str] = None
+    access_key_id: Optional[str] = None
+    secret_access_key: Optional[str] = None
+    region: Optional[str] = None
+    endpoint_url: Optional[str] = None
+    prefix: str = ""
+
+
 class S3Plugin(ConnectorPlugin):
     NAME = "s3"
     URI_SCHEME = "s3"
@@ -44,6 +54,7 @@ class S3Plugin(ConnectorPlugin):
         delete_detection="full_scan",
         paged_cat=True,
     )
+    CONFIG_SCHEMA = S3Config
 
     def _cfg(self, k, d=None):
         return (
@@ -142,8 +153,31 @@ async def read(self, path: str, range: Optional[Range] = None) -> AsyncIterator[
         key = path.lstrip("/")
         async with self._session().client("s3", **self._client_kwargs()) as s3:
             resp = await s3.get_object(Bucket=self._bucket(), Key=key)
-            async with resp["Body"] as stream:
-                yield await stream.read()
+            body = resp["Body"]
+            async with body:
+                # __aenter__ on the streaming body returns the raw aiohttp response (for
+                # connection lifecycle), not something with a chunk-sized read -- keep
+                # reading off `body` itself, which supports read(amt).
+                if range is None:
+                    while chunk := await body.read(65536):
+                        yield chunk
+                else:
+                    # line range [start, end) — stream line-by-line, same approach as the
+                    # file connector (cat --range must not buffer the whole object into memory).
+                    start, end = range.start, range.end
+                    i = 0
+                    buf = b""
+                    while chunk := await body.read(65536):
+                        buf += chunk
+                        while (nl := buf.find(b"\n")) >= 0:
+                            line, buf = buf[: nl + 1], buf[nl + 1 :]
+                            if start <= i < end:
+                                yield line
+                            i += 1
+                            if i >= end:
+                                return
+                    if buf and start <= i < end:  # trailing line without a newline
+                        yield buf
 
     async def fingerprint(self, path: str) -> Optional[str]:
         keys = await self.state.get("keys") or {}

server/python/src/mfs_server/connectors/web/plugin.py

@@ -15,6 +15,7 @@
 
 from ..base import (
     Capabilities,
+    ConnectorConfigSchema,
     ConnectorPlugin,
     Entry,
     HealthStatus,
@@ -26,6 +27,12 @@
 )
 
 
+class WebConfig(ConnectorConfigSchema):
+    start_urls: list[str] = []
+    allowed_domains: list[str] = []
+    max_pages: int = 50
+
+
 class WebPlugin(ConnectorPlugin):
     NAME = "web"
     URI_SCHEME = "web"
@@ -39,6 +46,7 @@ class WebPlugin(ConnectorPlugin):
         delete_detection="explicit",
         paged_cat=True,
     )
+    CONFIG_SCHEMA = WebConfig
 
     # Rendered page markdown, shared PROCESS-WIDE across plugin instances, keyed by
     # (connector_id, path). The in-process worker runs the index pass on a FRESH plugin

server/python/src/mfs_server/engine/components/connector_factory.py

@@ -21,8 +21,11 @@
 from dataclasses import replace as _replace
 from typing import Any
 
+from pydantic import ValidationError as _PydanticValidationError
+
 from ...config import ServerConfig
 from ...connectors.base import (
+    ConnectorConfigSchema,
     ConnectorContext,
     ConnectorPlugin,
     ObjectConfig,
@@ -97,6 +100,17 @@ def _match_object_config(objects_cfg: list, path: str) -> ObjectConfig | None:
     return None
 
 
+def _summarize_validation_error(e: _PydanticValidationError) -> str:
+    """Flatten pydantic's multi-line ValidationError into one clean,
+    grep-able line naming every offending field — pydantic's own str(e) is
+    verbose and spans several lines, awkward inside a single CLI error."""
+    parts = []
+    for err in e.errors():
+        loc = ".".join(str(x) for x in err["loc"]) or "(top-level)"
+        parts.append(f"{loc}: {err['msg']}")
+    return "; ".join(parts)
+
+
 # scheme extractor for target URIs; a bare path (no scheme) defaults to file.
 _SCHEME_RE = re.compile(r"^([a-z][a-z0-9+.\-]*)://")
 
@@ -424,6 +438,32 @@ def resolve_credential(self, value: Any) -> Any:
         PluginBuilder calls it internally."""
         return self._creds.resolve(value)
 
+    # --- config schema validation ---
+
+    def validate_config(self, ctype: str, config: dict) -> None:
+        """Validate a caller-supplied config against the connector's
+        CONFIG_SCHEMA (if it declares one), before any connection is
+        attempted. Callers registering/probing/estimating a connector with an
+        explicit config SHOULD call this — catches a wrong-typed value (e.g. a
+        quoted number) or an unrecognized field with one clean, named error
+        instead of a raw exception surfacing deep in the plugin's own
+        connect()/read path. A connector with no CONFIG_SCHEMA, or one that
+        predates ConnectorConfigSchema (FileConfig is a plain @dataclass, not
+        a pydantic model, and was never actually wired up to any enforcement
+        before this), is unaffected — opt-in per connector, not universal."""
+        if not isinstance(config, dict):
+            return
+        cls = get_plugin_cls(ctype)
+        schema = getattr(cls, "CONFIG_SCHEMA", None) if cls is not None else None
+        if schema is None or not (
+            isinstance(schema, type) and issubclass(schema, ConnectorConfigSchema)
+        ):
+            return
+        try:
+            schema(**config)
+        except _PydanticValidationError as e:
+            raise ValueError(f"config_invalid: {_summarize_validation_error(e)}") from e
+
     # --- plugin build ---
 
     def build_plugin(self, ctype: str, config: dict, connector_id: str) -> BuiltPlugin: