Neuroglancer short links

fileglancer/alembic/versions/2d1f0e6b8c91_add_neuroglancer_states_table.py

@@ -0,0 +1,42 @@
+"""add neuroglancer_states table
+
+Revision ID: 2d1f0e6b8c91
+Revises: 9812335c52b6
+Create Date: 2025-10-22 00:00:00.000000
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '2d1f0e6b8c91'
+down_revision = '9812335c52b6'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'neuroglancer_states',
+        sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column('short_key', sa.String(), nullable=False),
+        sa.Column('short_name', sa.String(), nullable=True),
+        sa.Column('username', sa.String(), nullable=False),
+        sa.Column('url_base', sa.String(), nullable=False),
+        sa.Column('state', sa.JSON(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), nullable=False),
+        sa.UniqueConstraint('short_key', name='uq_neuroglancer_states_short_key')
+    )
+    op.create_index(
+        'ix_neuroglancer_states_short_key',
+        'neuroglancer_states',
+        ['short_key'],
+        unique=True
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('ix_neuroglancer_states_short_key', table_name='neuroglancer_states')
+    op.drop_table('neuroglancer_states')

fileglancer/app.py

@@ -149,6 +149,38 @@ def _validate_filename(name: str) -> None:
         raise HTTPException(status_code=400, detail="File or directory name cannot have leading or trailing whitespace")
 
 
+def _parse_neuroglancer_url(url: str) -> Tuple[str, Dict]:
+    """
+    Parse a Neuroglancer URL and return its base URL and decoded JSON state.
+    """
+    if not url or "#!" not in url:
+        raise HTTPException(status_code=400, detail="Neuroglancer URL must include a '#!' state fragment")
+
+    url_base, encoded_state = url.split("#!", 1)
+    if not url_base.startswith(("http://", "https://")):
+        raise HTTPException(status_code=400, detail="Neuroglancer URL must start with http or https")
+
+    decoded_state = unquote(encoded_state)
+    if decoded_state.startswith(("http://", "https://")):
+        raise HTTPException(status_code=400, detail="Shortened Neuroglancer URLs are not supported; provide a full state URL")
+
+    try:
+        state = json.loads(decoded_state)
+    except json.JSONDecodeError:
+        raise HTTPException(status_code=400, detail="Neuroglancer state must be valid JSON")
+
+    if not isinstance(state, dict):
+        raise HTTPException(status_code=400, detail="Neuroglancer state must be a JSON object")
+
+    return url_base, state
+
+
+def _validate_short_name(short_name: str) -> None:
+    """Validate short_name: only letters, numbers, hyphens, and underscores allowed."""
+    if not all(ch.isalnum() or ch in ("-", "_") for ch in short_name):
+        raise HTTPException(status_code=400, detail="short_name can only contain letters, numbers, hyphens, and underscores")
+
+
 def create_app(settings):
 
     # Initialize OAuth client for OKTA
@@ -664,6 +696,119 @@ async def delete_preference(key: str, username: str = Depends(get_current_user))
             return {"message": f"Preference {key} deleted for user {username}"}
 
 
+    @app.post("/api/neuroglancer/nglinks", response_model=NeuroglancerShortenResponse,
+              description="Store a Neuroglancer state and return a shortened link")
+    async def shorten_neuroglancer_state(request: Request,
+                                         payload: NeuroglancerShortenRequest,
+                                         username: str = Depends(get_current_user)):
+        short_name = payload.short_name.strip() if payload.short_name else None
+        if short_name:
+            _validate_short_name(short_name)
+        title = payload.title.strip() if payload.title else None
+
+        if payload.url and payload.state:
+            raise HTTPException(status_code=400, detail="Provide either url or state, not both")
+
+        if payload.url:
+            url_base, state = _parse_neuroglancer_url(payload.url.strip())
+        elif payload.state:
+            if not payload.url_base:
+                raise HTTPException(status_code=400, detail="url_base is required when providing state directly")
+            if not isinstance(payload.state, dict):
+                raise HTTPException(status_code=400, detail="state must be a JSON object")
+            url_base = payload.url_base.strip()
+            if not url_base.startswith(("http://", "https://")):
+                raise HTTPException(status_code=400, detail="url_base must start with http or https")
+            state = payload.state
+        else:
+            raise HTTPException(status_code=400, detail="Either url or state must be provided")
+
+        # Add title to state if provided
+        if title:
+            state = {**state, "title": title}
+
+        with db.get_db_session(settings.db_url) as session:
+            try:
+                entry = db.create_neuroglancer_state(
+                    session,
+                    username,
+                    url_base,
+                    state,
+                    short_name=short_name
+                )
+                created_short_key = entry.short_key
+                created_short_name = entry.short_name
+            except ValueError as exc:
+                raise HTTPException(status_code=409, detail=str(exc))
+
+        # Generate URL based on whether short_name is provided
+        if created_short_name:
+            state_url = str(request.url_for("get_neuroglancer_state", short_key=created_short_key, short_name=created_short_name))
+        else:
+            state_url = str(request.url_for("get_neuroglancer_state_simple", short_key=created_short_key))
+        neuroglancer_url = f"{url_base}#!{state_url}"
+        return NeuroglancerShortenResponse(
+            short_key=created_short_key,
+            short_name=created_short_name,
+            title=title,
+            state_url=state_url,
+            neuroglancer_url=neuroglancer_url
+        )
+
+
+    @app.put("/api/neuroglancer/nglinks/{short_key}", response_model=NeuroglancerShortenResponse,
+             description="Update a stored Neuroglancer state")
+    async def update_neuroglancer_short_link(request: Request,
+                                             short_key: str,
+                                             payload: NeuroglancerUpdateRequest,
+                                             username: str = Depends(get_current_user)):
+        title = payload.title.strip() if payload.title else None
+        url_base, state = _parse_neuroglancer_url(payload.url.strip())
+
+        # Add title to state if provided
+        if title:
+            state = {**state, "title": title}
+
+        with db.get_db_session(settings.db_url) as session:
+            entry = db.update_neuroglancer_state(
+                session,
+                username,
+                short_key,
+                url_base,
+                state
+            )
+            if not entry:
+                raise HTTPException(status_code=404, detail="Neuroglancer state not found")
+            # Extract values before session closes
+            updated_short_key = entry.short_key
+            updated_short_name = entry.short_name
+
+        # Generate URL based on whether short_name is present
+        if updated_short_name:
+            state_url = str(request.url_for("get_neuroglancer_state", short_key=updated_short_key, short_name=updated_short_name))
+        else:
+            state_url = str(request.url_for("get_neuroglancer_state_simple", short_key=updated_short_key))
+        neuroglancer_url = f"{url_base}#!{state_url}"
+        return NeuroglancerShortenResponse(
+            short_key=updated_short_key,
+            short_name=updated_short_name,
+            title=title,
+            state_url=state_url,
+            neuroglancer_url=neuroglancer_url
+        )
+
+
+    @app.delete("/api/neuroglancer/nglinks/{short_key}",
+                description="Delete a stored Neuroglancer state")
+    async def delete_neuroglancer_short_link(short_key: str = Path(..., description="The short key of the Neuroglancer state"),
+                                             username: str = Depends(get_current_user)):
+        with db.get_db_session(settings.db_url) as session:
+            deleted = db.delete_neuroglancer_state(session, username, short_key)
+            if deleted == 0:
+                raise HTTPException(status_code=404, detail="Neuroglancer link not found")
+            return {"message": f"Neuroglancer link {short_key} deleted"}
+
+
     @app.post("/api/proxied-path", response_model=ProxiedPath,
               description="Create a new proxied path")
     async def create_proxied_path(fsp_name: str = Query(..., description="The name of the file share path that this proxied path is associated with"),
@@ -734,6 +879,59 @@ async def delete_proxied_path(sharing_key: str = Path(..., description="The shar
             return {"message": f"Proxied path {sharing_key} deleted for user {username}"}
 
 
+    @app.get("/ng/{short_key}", name="get_neuroglancer_state_simple", include_in_schema=False)
+    async def get_neuroglancer_state_simple(short_key: str = Path(..., description="Short key for a stored Neuroglancer state")):
+        with db.get_db_session(settings.db_url) as session:
+            entry = db.get_neuroglancer_state(session, short_key)
+            if not entry:
+                raise HTTPException(status_code=404, detail="Neuroglancer state not found")
+            # If this entry has a short_name, require it in the URL
+            if entry.short_name:
+                raise HTTPException(status_code=404, detail="Neuroglancer state not found")
+            return JSONResponse(content=entry.state, headers={"Cache-Control": "no-store"})
+
+    @app.get("/ng/{short_key}/{short_name}", name="get_neuroglancer_state", include_in_schema=False)
+    async def get_neuroglancer_state(short_key: str = Path(..., description="Short key for a stored Neuroglancer state"),
+                                     short_name: str = Path(..., description="Short name for a stored Neuroglancer state")):
+        with db.get_db_session(settings.db_url) as session:
+            entry = db.get_neuroglancer_state(session, short_key)
+            if not entry:
+                raise HTTPException(status_code=404, detail="Neuroglancer state not found")
+            # Validate short_name matches
+            if entry.short_name != short_name:
+                raise HTTPException(status_code=404, detail="Neuroglancer state not found")
+            return JSONResponse(content=entry.state, headers={"Cache-Control": "no-store"})
+
+
+    @app.get("/api/neuroglancer/nglinks", response_model=NeuroglancerShortLinkResponse,
+             description="List stored Neuroglancer short links for the current user")
+    async def get_neuroglancer_short_links(request: Request,
+                                           username: str = Depends(get_current_user)):
+        links = []
+        with db.get_db_session(settings.db_url) as session:
+            entries = db.get_neuroglancer_states(session, username)
+            for entry in entries:
+                # Generate URL based on whether short_name is provided
+                if entry.short_name:
+                    state_url = str(request.url_for("get_neuroglancer_state", short_key=entry.short_key, short_name=entry.short_name))
+                else:
+                    state_url = str(request.url_for("get_neuroglancer_state_simple", short_key=entry.short_key))
+                neuroglancer_url = f"{entry.url_base}#!{state_url}"
+                # Read title from the stored state
+                title = entry.state.get("title") if isinstance(entry.state, dict) else None
+                links.append(NeuroglancerShortLink(
+                    short_key=entry.short_key,
+                    short_name=entry.short_name,
+                    title=title,
+                    created_at=entry.created_at,
+                    updated_at=entry.updated_at,
+                    state_url=state_url,
+                    neuroglancer_url=neuroglancer_url
+                ))
+
+        return NeuroglancerShortLinkResponse(links=links)
+
+
     @app.get("/files/{sharing_key}/{sharing_name}")
     @app.get("/files/{sharing_key}/{sharing_name}/{path:path}")
     async def target_dispatcher(request: Request,

fileglancer/database.py

@@ -18,6 +18,7 @@
 
 # Constants
 SHARING_KEY_LENGTH = 12
+NEUROGLANCER_SHORT_KEY_LENGTH = 12
 
 # Global flag to track if migrations have been run
 _migrations_run = False
@@ -102,6 +103,20 @@ class ProxiedPathDB(Base):
     )
 
 
+class NeuroglancerStateDB(Base):
+    """Database model for storing Neuroglancer states"""
+    __tablename__ = 'neuroglancer_states'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    short_key = Column(String, nullable=False, unique=True, index=True)
+    short_name = Column(String, nullable=True)
+    username = Column(String, nullable=False)
+    url_base = Column(String, nullable=False)
+    state = Column(JSON, nullable=False)
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
+    updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
+
+
 class TicketDB(Base):
     """Database model for storing proxied paths"""
     __tablename__ = 'tickets'
@@ -614,6 +629,86 @@ def delete_proxied_path(session: Session, username: str, sharing_key: str):
     _invalidate_sharing_key_cache(sharing_key)
 
 
+def _generate_unique_neuroglancer_key(session: Session) -> str:
+    """Generate a unique short key for Neuroglancer states."""
+    for _ in range(10):
+        candidate = secrets.token_urlsafe(NEUROGLANCER_SHORT_KEY_LENGTH)
+        exists = session.query(NeuroglancerStateDB).filter_by(short_key=candidate).first()
+        if not exists:
+            return candidate
+    raise RuntimeError("Failed to generate a unique Neuroglancer short key")
+
+
+def create_neuroglancer_state(
+    session: Session,
+    username: str,
+    url_base: str,
+    state: Dict,
+    short_name: Optional[str] = None
+) -> NeuroglancerStateDB:
+    """Create a new Neuroglancer state entry and return it."""
+    short_key = _generate_unique_neuroglancer_key(session)
+    now = datetime.now(UTC)
+    entry = NeuroglancerStateDB(
+        short_key=short_key,
+        short_name=short_name,
+        username=username,
+        url_base=url_base,
+        state=state,
+        created_at=now,
+        updated_at=now
+    )
+    session.add(entry)
+    session.commit()
+    return entry
+
+
+def get_neuroglancer_state(session: Session, short_key: str) -> Optional[NeuroglancerStateDB]:
+    """Get a Neuroglancer state by short key."""
+    return session.query(NeuroglancerStateDB).filter_by(short_key=short_key).first()
+
+
+def get_neuroglancer_states(session: Session, username: str) -> List[NeuroglancerStateDB]:
+    """Get all Neuroglancer states for a user, newest first."""
+    return (
+        session.query(NeuroglancerStateDB)
+        .filter_by(username=username)
+        .order_by(NeuroglancerStateDB.created_at.desc())
+        .all()
+    )
+
+
+def update_neuroglancer_state(
+    session: Session,
+    username: str,
+    short_key: str,
+    url_base: str,
+    state: Dict
+) -> Optional[NeuroglancerStateDB]:
+    """Update a Neuroglancer state entry. Returns the updated entry or None if not found."""
+    entry = session.query(NeuroglancerStateDB).filter_by(
+        short_key=short_key,
+        username=username
+    ).first()
+    if not entry:
+        return None
+    entry.url_base = url_base
+    entry.state = state
+    entry.updated_at = datetime.now(UTC)
+    session.commit()
+    return entry
+
+
+def delete_neuroglancer_state(session: Session, username: str, short_key: str) -> int:
+    """Delete a Neuroglancer state entry. Returns the number of deleted rows."""
+    deleted = session.query(NeuroglancerStateDB).filter_by(
+        short_key=short_key,
+        username=username
+    ).delete()
+    session.commit()
+    return deleted
+
+
 def get_tickets(session: Session, username: str, fsp_name: str = None, path: str = None) -> List[TicketDB]:
     """Get tickets for a user, optionally filtered by fsp_name and path"""
     logger.info(f"Getting tickets for {username} with fsp_name={fsp_name} and path={path}")

fileglancer/filestore.py

@@ -195,8 +195,13 @@ def _get_file_info_from_path(self, full_path: str, current_user: str = None) ->
         Get the FileInfo for a file or directory at the given path.
         """
         stat_result = os.stat(full_path)
-        # Regenerate the relative path to ensure it is not empty (None and empty string are converted to '.' here)
-        rel_path = os.path.relpath(full_path, self.root_path)
+        # Use real paths to avoid /var vs /private/var mismatches on macOS.
+        root_real = os.path.realpath(self.root_path)
+        full_real = os.path.realpath(full_path)
+        if full_real == root_real:
+            rel_path = '.'
+        else:
+            rel_path = os.path.relpath(full_real, root_real)
         return FileInfo.from_stat(rel_path, full_path, stat_result, current_user)