Merge pull request #53 from SolidLabResearch/fix/source-pod-authentication

Merge protected source-Pod authentication review fixes.

Author: argahsuknesib

README.md

@@ -136,13 +136,13 @@ Send a JSON message containing a query, a processing type, and a client identifi
 
 ## Credentials
 
-### Normal live operation
+Heimdall can access both public and protected Solid streams. Authentication is a property of the source Pod or stream deployment, not of whether a query is `live` or `historical+live`.
 
-Normal `live` operation, including the Sensors-style live evaluation path, does not load source-Pod client credentials.
+For a protected source, copy [source-pod-credentials.example.json](./config/source-pod-credentials.example.json) to `config/source-pod-credentials.local.json`, fill it locally, and keep it untracked. Alternatively, set `HEIMDALL_SOURCE_POD_CREDENTIALS_FILE` to a local credential-file path. Each key is a source stream URL or a Pod URL prefix; matching is restricted to the same origin and path boundary, and the most-specific matching entry is selected. Same-origin resources discovered from a configured stream can reuse that source session; cross-origin notification resources require their own matching configuration. The entry contains a CSS client-credential `id`, `secret`, and `idp`.
 
-### historical+live
+When configured, Heimdall reuses the resulting authenticated session for relevant source operations: Type Index lookup, LDES metadata and historical retrieval, notification inbox and subscription-server discovery, notification subscription creation, and notification-event retrieval. When no matching source credentials are configured, these operations use ordinary unauthenticated HTTP, so public streams do not need a credentials file. A protected resource with no usable credentials reports the underlying authorization failure; an invalid configured entry reports a configuration error.
 
-`historical+live` requires a client-credential entry for each source stream. Copy [source-pod-credentials.example.json](./config/source-pod-credentials.example.json) to `config/source-pod-credentials.local.json`, fill it locally, and keep it untracked. Alternatively, set `HEIMDALL_SOURCE_POD_CREDENTIALS_FILE` to a local credential-file path.
+This does not assert that the Sensors evaluation used authentication; its deployment-specific authorization setup is not established by this repository.
 
 ### Legacy aggregation-Pod functionality
 

src/config/privateCredentials.test.ts

@@ -1,7 +1,7 @@
 import * as fs from 'fs';
 import * as os from 'os';
 import * as path from 'path';
-import { aggregationPodAccountFile, loadAggregationPodCredentials, loadSourcePodCredentials } from './privateCredentials';
+import { aggregationPodAccountFile, loadAggregationPodCredentials, loadOptionalSourcePodCredentials, loadSourcePodCredentials } from './privateCredentials';
 
 describe('private credential configuration', () => {
     const previousSource = process.env.HEIMDALL_SOURCE_POD_CREDENTIALS_FILE;
@@ -36,4 +36,18 @@ describe('private credential configuration', () => {
         delete process.env.HEIMDALL_SOURCE_POD_CREDENTIALS_FILE;
         expect(() => loadSourcePodCredentials()).toThrow('Missing source-Pod credentials');
     });
+
+    it('treats absent source credentials as an optional public-stream configuration', () => {
+        delete process.env.HEIMDALL_SOURCE_POD_CREDENTIALS_FILE;
+        expect(loadOptionalSourcePodCredentials()).toEqual({});
+    });
+
+    it('fails clearly for an explicitly missing or malformed source credential file', () => {
+        process.env.HEIMDALL_SOURCE_POD_CREDENTIALS_FILE = path.join(os.tmpdir(), 'does-not-exist-heimdall-source.json');
+        expect(() => loadOptionalSourcePodCredentials()).toThrow('Missing source-Pod credentials file');
+        const malformedPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'heimdall-credentials-')), 'malformed.json');
+        fs.writeFileSync(malformedPath, '{not-json');
+        process.env.HEIMDALL_SOURCE_POD_CREDENTIALS_FILE = malformedPath;
+        expect(() => loadOptionalSourcePodCredentials()).toThrow('Unable to load source-Pod credentials');
+    });
 });

src/config/privateCredentials.ts

@@ -16,6 +16,24 @@ export function loadSourcePodCredentials(): SourcePodCredentials {
     return loadJson('HEIMDALL_SOURCE_POD_CREDENTIALS_FILE', 'config/source-pod-credentials.local.json', 'source-Pod credentials');
 }
 
+/**
+ * Source-Pod authentication is optional: public streams must not require a
+ * local credential file. Invalid configured files still fail loudly.
+ */
+export function loadOptionalSourcePodCredentials(): SourcePodCredentials {
+    const configuredPath = process.env.HEIMDALL_SOURCE_POD_CREDENTIALS_FILE;
+    const filePath = configuredPath || path.resolve(process.cwd(), 'config/source-pod-credentials.local.json');
+    if (!fs.existsSync(filePath)) {
+        if (configuredPath) throw new Error(`Missing source-Pod credentials file configured by HEIMDALL_SOURCE_POD_CREDENTIALS_FILE: ${filePath}`);
+        return {};
+    }
+    try {
+        return JSON.parse(fs.readFileSync(filePath, 'utf8')) as SourcePodCredentials;
+    } catch (error) {
+        throw new Error(`Unable to load source-Pod credentials from ${filePath}: ${(error as Error).message}`);
+    }
+}
+
 export function loadAggregationPodCredentials(): AggregationPodCredentials {
     return loadJson('HEIMDALL_AGGREGATION_POD_CREDENTIALS_FILE', 'config/aggregation-pod-credentials.local.json', 'aggregation-Pod credentials');
 }

src/server/WebSocketHandler.test.ts

@@ -210,7 +210,7 @@ WHERE {
 
         expect(result.ldes_query).toContain(`<${streamUrl}>`);
         expect(result.ldes_query).not.toContain('undefined');
-        expect(discovery).toHaveBeenCalledWith(podUrl, expect.any(Array));
+        expect(discovery).toHaveBeenCalledWith(podUrl, expect.any(Array), undefined, expect.anything());
         const metrics = fs.readFileSync(path.join((handler as any).metric_writer.resultsDir, 'initialization.csv'), 'utf8');
         expect(metrics).toContain('stream_discovery');
         discovery.mockRestore();

src/server/WebSocketHandler.ts

@@ -19,6 +19,7 @@ import { MetricWriter } from '../evaluation/MetricWriter';
 import { createHash } from 'crypto';
 import { SharedStreamRegistry } from '../service/heimdall/SharedStreamRegistry';
 import { loadAggregationPodCredentials } from '../config/privateCredentials';
+import { SourcePodAccess } from '../service/heimdall/SourcePodAccess';
 
 /**
  * Class for handling the Websocket server.
@@ -37,6 +38,7 @@ export class WebSocketHandler {
     public logger: any;
     private query_registry: QueryRegistry;
     private readonly metric_writer: MetricWriter;
+    private readonly sourcePodAccess: SourcePodAccess;
     private aggregationPublisherRegistered = false;
     /**
      * Creates an instance of WebSocketHandler.
@@ -55,6 +57,7 @@ export class WebSocketHandler {
         this.connections = new Map<string, WebSocket[]>();
         this.parser = new RSPQLParser();
         this.metric_writer = metric_writer;
+        this.sourcePodAccess = sharedStreamRegistry?.sourcePodAccess || new SourcePodAccess();
         this.query_registry = new QueryRegistry(metric_writer, sharedStreamRegistry);
         this.n3_parser = new Parser({ format: 'N-Triples' });
         this.logger.info({}, 'websocket_handler_initialized');
@@ -333,7 +336,7 @@ export class WebSocketHandler {
         const interest_metric = new AggregationFocusExtractor(query).extract_focus();
         const start_epoch_ms = Date.now();
         const start_monotonic_ns = process.hrtime.bigint();
-        const streams = await find_relevant_streams(pod_url, interest_metric);
+        const streams = await find_relevant_streams(pod_url, interest_metric, undefined, this.sourcePodAccess);
         const ldes_stream = streams[0];
         if (ldes_stream === undefined) {
             throw new Error(`No relevant LDES stream found for Pod source ${pod_url}`);

src/service/heimdall/DecentralizedFileStreamer.ts

@@ -8,13 +8,12 @@ const websocketConnection = require('websocket').connection;
 const WebSocketClient = require('websocket').client;
 import { Quad } from "n3";
 import { QuadWithID } from "../../utils/Types";
-import { session_with_credentials } from "../../utils/authentication/CSSAuthentication";
 import { readMembersRateLimited } from "../../utils/ldes-in-ldp/EventSource";
 import { RateLimitedLDPCommunication } from "rate-limited-ldp-communication";
 import { hash_string_md5 } from "../../utils/Util";
 import { TREE } from "@treecg/ldes-snapshot";
-import { Session } from "@inrupt/solid-client-authn-node";
 import { create_subscription, extract_ldp_inbox, extract_subscription_server } from "../../utils/notifications/Util";
+import { SourcePodAccess } from './SourcePodAccess';
 import * as HEIMDALL_SETUP from '../../config/heimdall_setup.json';
 import { resolveHeimdallSetupConfig, resolveHeimdallWebSocketUrl } from "../../config/heimdallConfig";
 import { HEIMDALL_WEBSOCKET_PROTOCOL } from "../../server/websocketProtocols";
@@ -32,7 +31,7 @@ export class DecentralizedFileStreamer {
     public ldes!: LDESinLDP;
     public comunica_engine: QueryEngine;
     public communication: Promise<SolidCommunication | LDPCommunication | RateLimitedLDPCommunication>;
-    public session: any;
+    private readonly sourcePodAccess: SourcePodAccess;
     public observation_array: any[];
     public query: string
     public query_hash: string;
@@ -43,17 +42,18 @@ export class DecentralizedFileStreamer {
     /**
      * Creates an instance of DecentralizedFileStreamer.
      * @param {string} ldes_stream - The LDES stream URL.
-     * @param {session_credentials} session_credentials - The credentials of the Solid Pod.
+     * @param {SourcePodAccess} sourcePodAccess - Shared optional authenticated access for the source Pod.
      * @param {Date} from_date - The start date of the events to be read from the Solid Pod.
      * @param {Date} to_date - The end date of the events to be read from the Solid Pod.
      * @param {RSPEngine} rsp_engine - The RSP Engine.
      * @param {string} query - The query to be executed.
      * @param {*} logger - The logger object.
      * @memberof DecentralizedFileStreamer
      */
-    constructor(ldes_stream: string, session_credentials: session_credentials, from_date: Date, to_date: Date, rsp_engine: RSPEngine, query: string, logger: any) {
+    constructor(ldes_stream: string, sourcePodAccess: SourcePodAccess, from_date: Date, to_date: Date, rsp_engine: RSPEngine, query: string, logger: any) {
         this.ldes_stream = ldes_stream;
-        this.communication = this.get_communication(session_credentials);
+        this.sourcePodAccess = sourcePodAccess;
+        this.communication = this.sourcePodAccess.communicationFor(ldes_stream);
         this.from_date = from_date;
         this.to_date = to_date;
         this.query = query;
@@ -63,7 +63,6 @@ export class DecentralizedFileStreamer {
         this.stream_name = rsp_engine.getStream(this.ldes_stream);
         this.comunica_engine = new QueryEngine();
         this.observation_array = [];
-        this.subscribing_latest_events(this.stream_name);
         DecentralizedFileStreamer.connect_with_server(resolveHeimdallWebSocketUrl(HEIMDALL_SETUP)).then(() => {
             console.log(`The connection with the websocket server was established.`);
         });
@@ -78,16 +77,6 @@ export class DecentralizedFileStreamer {
      * @returns {Promise<SolidCommunication | LDPCommunication>} - The communication object with the Solid Pod.
      * @memberof DecentralizedFileStreamer
      */
-    public async get_communication(credentials: session_credentials) {
-        const session = await this.get_session(credentials);
-        if (session) {
-            return new SolidCommunication(session);
-        }
-        else {
-            return new LDPCommunication();
-        }
-    }
-
     /**
      * Adding the events which might have been added between 
      * the start of the file streamer and the start of the websocket
@@ -240,12 +229,13 @@ export class DecentralizedFileStreamer {
     async subscribing_latest_events(stream_name: RDFStream | undefined) {
         if (stream_name !== undefined) {
             console.log(`Subscribing to the latest events of the stream`, stream_name);
-            const inbox = await extract_ldp_inbox(this.ldes_stream);
+            const sourceFetch = await this.sourcePodAccess.fetchFor(this.ldes_stream);
+            const inbox = await extract_ldp_inbox(this.ldes_stream, undefined, sourceFetch);
             if (inbox !== undefined) {
-                const subscription_server = await extract_subscription_server(inbox);
+                const subscription_server = await extract_subscription_server(inbox, undefined, await this.sourcePodAccess.fetchFor(inbox, this.ldes_stream));
                 if (subscription_server !== undefined) {
                     const server = subscription_server.location;
-                    const response_subscription = await create_subscription(server, inbox);
+                    const response_subscription = await create_subscription(server, inbox, undefined, await this.sourcePodAccess.fetchFor(server, this.ldes_stream));
                     if (response_subscription) {
                         this.logger.info(`The subscription has been succesful.`);
                     } else {
@@ -274,7 +264,7 @@ export class DecentralizedFileStreamer {
      */
     async get_inbox_container(stream: string): Promise<string | undefined> {
         console.log(`Getting the inbox container from`, stream);
-        const ldes_in_ldp: LDESinLDP = new LDESinLDP(stream, new LDPCommunication());
+        const ldes_in_ldp: LDESinLDP = new LDESinLDP(stream, await this.sourcePodAccess.communicationFor(stream));
         const metadata = await ldes_in_ldp.readMetadata();
         for (const quad of metadata) {
             if (quad.predicate.value === 'http://www.w3.org/ns/ldp#inbox') {
@@ -308,7 +298,7 @@ export class DecentralizedFileStreamer {
             "sendTo": `${heimdallSetupConfig.heimdallHttpServerUrl}`
         };
 
-        const response = await fetch(webhook_notification_server, {
+        const response = await (await this.sourcePodAccess.fetchFor(webhook_notification_server, ldes_stream))(webhook_notification_server, {
             method: 'POST',
             headers: {
                 'Content-Type': 'application/ld+json',
@@ -335,7 +325,7 @@ export class DecentralizedFileStreamer {
             "type": "http://www.w3.org/ns/solid/notifications#WebSocketChannel2023",
             "topic": `${ldes_stream}`
         }
-        const repsonse = await fetch(notification_server, {
+        const repsonse = await (await this.sourcePodAccess.fetchFor(notification_server, ldes_stream))(notification_server, {
             method: 'POST',
             headers: {
                 'Content-Type': 'application/ld+json',
@@ -379,7 +369,7 @@ export class DecentralizedFileStreamer {
             "type": "http://www.w3.org/ns/solid/notifications#WebSocketChannel2023",
             "topic": `${inbox_container}`
         }
-        const repsonse = await fetch(notification_server, {
+        const repsonse = await (await this.sourcePodAccess.fetchFor(notification_server, ldes_stream))(notification_server, {
             method: 'POST',
             headers: {
                 'Content-Type': 'application/ld+json',
@@ -408,15 +398,6 @@ export class DecentralizedFileStreamer {
     get_file_streamer_start_time() {
         return this.file_streamer_start_time;
     }
-    /**
-     * Get the session with the credentials.
-     * @param {session_credentials} credentials - The credentials of the solid pod for which you can generate an authenticated session to communicated to the Solid Pod's LDP.
-     * @returns {Promise<Session>} - The authenticated session.
-     * @memberof DecentralizedFileStreamer
-     */
-    async get_session(credentials: session_credentials): Promise<Session> {
-        return await session_with_credentials(credentials);
-    }
     /**
      * Send a message to Heimdall's WebSocket server.
      * @static