Merge pull request #590 from desci-labs/m0ar/fix-stream-checks

Fix token ownership checks on upgrade, et. al.

desci-contracts/scripts/alias-registry/manualUpgrade.mjs

@@ -0,0 +1,73 @@
+/**
+ * MANUAL DPID UPGRADE
+ *
+ * Manually upgrade a legacy dPID to an alias, e.g., bind a streamID
+ * to a dPID in `registry` and `reverseRegistry` mappings.
+ *
+ * Notes:
+ * - This prevents (unpriviligied) binding of this stream ID to another dPID
+ *
+ * Required arguments (env variables):
+ * 1. ENV - dev or prod
+ * 2. REGISTRY_ADDRESS - Address of existing alias registry (proxy) contract
+ * 3. PRIVATE_KEY - Owner/admin identity (see hardhat.config.ts)
+ * 4. DPID - The dPID to bind
+ * 5. STREAM_ID - The streamID to bind
+ * 6. CONFIRM - Set "yes" to actually execute, otherwise run only checks
+ */
+import hardhat from "hardhat";
+const { ethers } = hardhat;
+
+const ENV = process.env.ENV;
+if (!(ENV === "dev" || ENV === "prod")) {
+  throw new Error('ENV unset (wanted "dev" or "prod")');
+};
+
+const CERAMIC_API = `https://ceramic-${ENV}.desci.com/api/v0/streams/`;
+
+const REGISTRY_ADDRESS = process.env.REGISTRY_ADDRESS;
+if (!REGISTRY_ADDRESS) {
+  throw new Error("REGISTRY_ADDRESS unset");
+};
+
+const DPID = process.env.DPID;
+if (!DPID) {
+  throw new Error("DPID unset");
+};
+
+const STREAM_ID = process.env.STREAM_ID;
+if (!STREAM_ID) {
+  throw new Error("STREAM_ID unset");
+};
+
+const DpidAliasRegistryFactory = await ethers.getContractFactory("DpidAliasRegistry");
+const registry = DpidAliasRegistryFactory.attach(REGISTRY_ADDRESS);
+
+const dpidLookup = await registry.resolve(DPID);
+const freeDpid = dpidLookup === "";
+console.log(`➡ dPID ${DPID} unbound: ${freeDpid ? "✅" : "❌"}`);
+
+const reverseLookup = await registry.find(STREAM_ID);
+const freeStreamID = reverseLookup.toNumber() === 0;
+console.log(`➡ Stream ${STREAM_ID} unbound: ${freeStreamID ? "✅" : "❌"}`);
+
+const [legacyOwner, _versions ] = await registry.legacyLookup(DPID);
+const res = await fetch(CERAMIC_API + STREAM_ID);
+const body = await res.json();
+const streamController = res.ok
+  ? body.state.metadata.controllers[0]
+  : "UNKNOWN";
+const sameOwner = legacyOwner.toLowerCase() === streamController.split(":").pop();
+console.log(
+  `➡ Same owner: ${sameOwner ? "✅" : "❌"}`,
+  { legacyOwner, streamController }
+);
+
+console.log();
+if (process.env.CONFIRM === "yes") {
+  const setNextDpid = await registry.upgradeDpid(NEXT_DPID)
+  await setNextDpid.wait();
+  console.log(`🆙 Bound ${DPID} to ${STREAM_ID}`);
+} else {
+  console.log(`🙅 Skipping binding ${DPID} to ${STREAM_ID} (set CONFIRM to execute)`);
+}

desci-media-isolated/Dockerfile

@@ -1,5 +1,5 @@
 
-FROM docker.io/node:20.9.0 as base
+FROM docker.io/node:20.9.0 AS base
 
 # Install dumb-init so we can use it as PID 1
 
@@ -16,7 +16,7 @@ WORKDIR /usr/src/app
 COPY tsconfig.json .
 COPY package*.json ./
 
-FROM base as dev
+FROM base AS dev
 
 RUN --mount=type=cache,target=/usr/src/app/.npm \
   npm set cache /usr/src/app/.npm && \
@@ -32,7 +32,7 @@ EXPOSE 9777
 ENTRYPOINT ["/usr/src/app/scripts/containerInitDev.sh"]
 CMD ["dumb-init", "npx", "tsx","watch", "--clear-screen=false", "--env-file=.env", "--inspect=0.0.0.0:9777", "src/index.ts"]
 
-FROM base as production
+FROM base AS production
 # Cache mounts for faster builds, prod env for better express perf
 RUN --mount=type=cache,target=/usr/src/app/.npm \
   npm set cache /usr/src/app/.npm && \

desci-repo/Dockerfile

@@ -9,9 +9,10 @@ WORKDIR /app
 RUN mkdir /app/repo-tmp
 RUN mkdir -p /app/desci-repo/repo-tmp
 
-COPY . .
-
+COPY package.json .
 RUN --mount=type=cache,target=/root/.yarn YARN_CACHE_FOLDER=/root/.yarn yarn install
+
+COPY . .
 RUN yarn build
 
 # server api
@@ -20,4 +21,4 @@ EXPOSE 5484
 # websocket
 EXPOSE 5445
 
-CMD [ "yarn", "start" ]
+CMD [ "yarn", "start" ]

desci-server/package.json

@@ -42,7 +42,6 @@
     "test": "yarn test:check && yarn docker:test; export EXIT=$(echo $?); docker compose  --file ../docker-compose.test.yml --compatibility down; exit $EXIT",
     "coverage:destructive": "nyc --all --parser-plugins='[\"importAssertions\"]' -r lcov -e .ts -x \"*.test.ts\" npm run test:destructive",
     "coverage:destructive:debug": "nyc --all --parser-plugins='[\"importAssertions\"]' -r lcov -e .ts -x \"*.test.ts\" npm run test:destructive:debug",
-    "commit": "git-cz",
     "db:forward": "kubectl run --env REMOTE_HOST=$REMOTE_HOST --env REMOTE_PORT=5432 --env LOCAL_PORT=8080 --port 8080 --image marcnuri/port-forward test-port-forward ;  kubectl port-forward test-port-forward 8080:8080",
     "docker:dev": "../dockerDev.sh",
     "docker:test": "CI=true docker compose --file ../docker-compose.test.yml --compatibility up --exit-code-from nodes_backend_test",
@@ -112,6 +111,7 @@
     "pino": "^8.14.1",
     "pino-http": "^8.3.3",
     "pino-pretty": "^10.0.0",
+    "pino-std-serializers": "^7.0.0",
     "prisma": "4.10.1",
     "qs": "^6.12.0",
     "react-email": "2.1.0",
@@ -132,8 +132,6 @@
     "zod": "^3.22.4"
   },
   "devDependencies": {
-    "@commitlint/cli": "^13.1.0",
-    "@commitlint/config-conventional": "^13.1.0",
     "@types/archiver": "^6.0.2",
     "@types/body-parser": "^1.19.1",
     "@types/chai": "^4.2.21",
@@ -150,7 +148,6 @@
     "@typescript-eslint/eslint-plugin": "^6.14.0",
     "@typescript-eslint/parser": "^6.14.0",
     "chai": "^4.3.4",
-    "commitizen": "^4.2.4",
     "copyfiles": "^2.4.1",
     "dotenv": "^10.0.0",
     "eslint": "^7.32.0",
@@ -176,17 +173,7 @@
       "eslint --max-warnings 0"
     ]
   },
-  "commitlint": {
-    "extends": [
-      "@commitlint/config-conventional"
-    ]
-  },
-  "config": {
-    "commitizen": {
-      "path": "node_modules/cz-conventional-changelog"
-    }
-  },
   "prisma": {
     "seed": "node --no-warnings=ExperimentalWarning --loader ts-node/esm prisma/seed.ts"
   }
-}
+}

desci-server/src/controllers/data/retrieve.ts

@@ -10,10 +10,10 @@ import tar from 'tar';
 
 import { prisma } from '../../client.js';
 import { logger as parentLogger } from '../../logger.js';
-import redisClient, { getOrCache } from '../../redisClient.js';
+import { redisClient, getOrCache } from '../../redisClient.js';
 import { getLatestDriveTime } from '../../services/draftTrees.js';
 import { getDatasetTar } from '../../services/ipfs.js';
-import { NodeUuid, getLatestManifestFromNode } from '../../services/manifestRepo.js';
+import { NodeUuid } from '../../services/manifestRepo.js';
 import { showNodeDraftManifest } from '../../services/nodeManager.js';
 import { getTreeAndFill, getTreeAndFillDeprecated } from '../../utils/driveUtils.js';
 import { cleanupManifestUrl } from '../../utils/manifest.js';

desci-server/src/controllers/nodes/draftCreate.ts

@@ -146,7 +146,8 @@ export const draftCreate = async (req: Request, res: Response, next: NextFunctio
     });
 
     // cache initial doc for a minute (60)
-    await setToCache(`node-draft-${node.uuid}`, { document, documentId }, 60);
+    // ! disabling, as it breaks programmatic interaction from nodes-lib, where stale results break interactivity
+    // await setToCache(`node-draft-${ensureUuidEndsWithDot(node.uuid)}`, { document, documentId }, 60);
 
     return;
   } catch (err) {

desci-server/src/controllers/nodes/prepublish.ts

@@ -4,12 +4,14 @@ import { NextFunction, Response } from 'express';
 
 import { prisma } from '../../client.js';
 import { logger as parentLogger } from '../../logger.js';
-import { RequestWithNode } from '../../middleware/authorisation.js';
+import { ensureNodeAccess, RequestWithNode } from '../../middleware/authorisation.js';
+import { delFromCache } from '../../redisClient.js';
 import { updateManifestDataBucket } from '../../services/data/processing.js';
 import { NodeUuid } from '../../services/manifestRepo.js';
 import repoService from '../../services/repoService.js';
 import { prepareDataRefsForDagSkeleton } from '../../utils/dataRefTools.js';
 import { dagifyAndAddDbTreeToIpfs } from '../../utils/draftTreeUtils.js';
+import { ensureUuidEndsWithDot } from '../../utils.js';
 import { persistManifest } from '../data/utils.js';
 
 type PrepublishResponse = PrepublishSuccessResponse | PrepublishErrorResponse;
@@ -45,6 +47,7 @@ export const prepublish = async (req: RequestWithNode, res: Response<PrepublishR
   if (!uuid) {
     return res.status(400).json({ ok: false, error: 'UUID is required.' });
   }
+
   try {
     // Sourced from middleware EnsureUser
     if (!owner.id || owner.id < 1) {

desci-server/src/controllers/nodes/publish.ts

@@ -1,10 +1,13 @@
 import { ResearchObjectV1 } from '@desci-labs/desci-models';
 import { ActionType, Node, Prisma, PublishTaskQueue, PublishTaskQueueStatus, User } from '@prisma/client';
 import { Request, Response, NextFunction } from 'express';
+import { stdSerializers } from 'pino';
 
 import { prisma } from '../../client.js';
 import { logger as parentLogger } from '../../logger.js';
+import { delFromCache } from '../../redisClient.js';
 import { attestationService } from '../../services/Attestation.js';
+import { directStreamLookup } from '../../services/ceramic.js';
 import { getManifestByCid } from '../../services/data/processing.js';
 import { getTargetDpidUrl } from '../../services/fixDpid.js';
 import { saveInteraction, saveInteractionWithoutReq } from '../../services/interactionLog.js';
@@ -17,7 +20,7 @@ import {
 } from '../../services/nodeManager.js';
 import { emitNotificationOnPublish } from '../../services/NotificationService.js';
 import { publishServices } from '../../services/PublishServices.js';
-import { getIndexedResearchObjects } from '../../theGraph.js';
+import { _getIndexedResearchObjects, getIndexedResearchObjects } from '../../theGraph.js';
 import { discordNotify } from '../../utils/discordUtils.js';
 import { ensureUuidEndsWithDot } from '../../utils.js';
 
@@ -61,10 +64,9 @@ async function updateAssociatedAttestations(nodeUuid: string, dpid: string) {
     },
   });
 }
-// call node publish service and add job to queue
+
 export const publish = async (req: PublishRequest, res: Response<PublishResBody>, _next: NextFunction) => {
   const { uuid, cid, manifest, transactionId, ceramicStream, commitId, useNewPublish } = req.body;
-  // debugger;
   const email = req.user.email;
   const logger = parentLogger.child({
     // id: req.id,
@@ -124,40 +126,8 @@ export const publish = async (req: PublishRequest, res: Response<PublishResBody>
 
     if (task) return res.status(400).json({ error: 'Node publishing in progress' });
 
-    // let publishTask: PublishTaskQueue | undefined;
     logger.info({ ceramicStream, commitId, uuid, owner: owner.id }, 'Triggering new publish flow');
     const dpidAlias = await syncPublish(ceramicStream, commitId, node, owner, cid, uuid, manifest);
-    // if (useNewPublish) {
-    // } else {
-    // publishTask = await prisma.publishTaskQueue.create({
-    //   data: {
-    //     cid,
-    //     dpid: manifest.dpid?.id,
-    //     userId: owner.id,
-    //     transactionId,
-    //     ceramicStream,
-    //     commitId,
-    //     uuid: ensureUuidEndsWithDot(uuid),
-    //     status: PublishTaskQueueStatus.WAITING,
-    //   },
-    // });
-    // }
-
-    saveInteraction(
-      req,
-      ActionType.PUBLISH_NODE,
-      {
-        cid,
-        dpid: dpidAlias?.toString() ?? manifest.dpid?.id,
-        userId: owner.id,
-        transactionId,
-        ceramicStream,
-        commitId,
-        uuid: ensureUuidEndsWithDot(uuid),
-        // status: PublishTaskQueueStatus.WAITING,
-      },
-      owner.id,
-    );
 
     updateAssociatedAttestations(node.uuid, dpidAlias ? dpidAlias.toString() : manifest.dpid?.id);
 
@@ -179,13 +149,41 @@ export const publish = async (req: PublishRequest, res: Response<PublishResBody>
       version,
     });
 
+    // Make sure we don't serve stale manifest state when a publish is happening
+    delFromCache(`node-draft-${ensureUuidEndsWithDot(node.uuid)}`);
+
+    saveInteraction(
+      req,
+      ActionType.PUBLISH_NODE,
+      {
+        cid,
+        dpid: dpidAlias?.toString() ?? manifest.dpid?.id,
+        userId: owner.id,
+        transactionId,
+        ceramicStream,
+        commitId,
+        uuid: ensureUuidEndsWithDot(uuid),
+        outcome: 'SUCCESS',
+      },
+      owner.id,
+    );
+
     return res.send({
       ok: true,
       dpid: dpidAlias ?? parseInt(manifest.dpid?.id),
-      // taskId: publishTask?.id,
     });
   } catch (err) {
     logger.error({ err }, '[publish::publish] node-publish-err');
+    saveInteraction(req, ActionType.PUBLISH_NODE, {
+      cid,
+      user: req.user,
+      transactionId,
+      ceramicStream,
+      commitId,
+      uuid: ensureUuidEndsWithDot(uuid),
+      outcome: 'FAILURE',
+      err: stdSerializers.errWithCause(err as Error),
+    });
     return res.status(400).send({ ok: false, error: err.message });
   }
 };
@@ -313,8 +311,39 @@ const createOrUpgradeDpidAlias = async (
   ceramicStream: string,
   uuid: string,
 ): Promise<number> => {
+  const logger = parentLogger.child({
+    module: 'NODE::createOrUpgradeDpidAlias',
+    legacyDpid,
+    ceramicStream,
+    uuid,
+  });
+
   let dpidAlias: number;
   if (legacyDpid) {
+    // Use subgraph lookup to ensure we don't get the owner from the stream and compare with itself
+    const legacyHistory = await _getIndexedResearchObjects([uuid]);
+
+    // On the initial legacy publish, the subgraph hasn't had time to index the event at this point.
+    // If it returns successfully, but array is empty, we can assume this is the first publish.
+    // and OK the check as there isn't any older history to contend with.
+    // This likely only happens in the legacy publish tests in nodes-lib, as legacy publish is disabled in the app.
+    const legacyOwner = legacyHistory.researchObjects[0]?.owner;
+
+    const streamInfo = await directStreamLookup(ceramicStream);
+    if ('err' in streamInfo) {
+      logger.error(streamInfo, 'Failed to load stream when doing checks before upgrade');
+      throw new Error('Failed to load stream');
+    }
+    const streamController = streamInfo.state.metadata.controllers[0].toLowerCase();
+    const differentOwner = legacyOwner?.toLowerCase() !== streamController.split(':').pop().toLowerCase();
+
+    // Caveat from above: if there was a legacyDpid, but no owner, we're likely in the middle of that process
+    // and nodes-lib has published both with the same key regardless
+    if (differentOwner && legacyOwner !== undefined) {
+      logger.error({ streamController, legacyOwner }, 'Legacy owner and stream controller differs');
+      throw new Error('Legacy owner and stream controller differs');
+    }
+
     // Requires the REGISTRY_OWNER_PKEY to be set in env
     dpidAlias = await upgradeDpid(legacyDpid, ceramicStream);
   } else {