fix: authenticate protected source Pod access

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; the most-specific matching entry is selected for that source. 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,9 @@ 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({});
+    });
 });

src/config/privateCredentials.ts

@@ -16,6 +16,15 @@ 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 filePath = process.env.HEIMDALL_SOURCE_POD_CREDENTIALS_FILE || path.resolve(process.cwd(), 'config/source-pod-credentials.local.json');
+    return fs.existsSync(filePath) ? JSON.parse(fs.readFileSync(filePath, 'utf8')) as SourcePodCredentials : {};
+}
+
 export function loadAggregationPodCredentials(): AggregationPodCredentials {
     return loadJson('HEIMDALL_AGGREGATION_POD_CREDENTIALS_FILE', 'config/aggregation-pod-credentials.local.json', 'aggregation-Pod credentials');
 }

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;
@@ -78,16 +78,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 +230,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, sourceFetch);
                 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, sourceFetch);
                     if (response_subscription) {
                         this.logger.info(`The subscription has been succesful.`);
                     } else {
@@ -408,15 +399,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

src/service/heimdall/HeimdallInstantiator.ts

@@ -5,14 +5,14 @@ import { v4 as uuidv4 } from 'uuid';
 import { EventEmitter } from "events";
 import { BindingsWithTimestamp } from "../../utils/Types";
 import { hash_string_md5 } from "../../utils/Util";
-import { Credentials, aggregation_object } from "../../utils/Types";
+import { aggregation_object } from "../../utils/Types";
 import { SharedStreamRegistry } from './SharedStreamRegistry';
 import { N3ReasonerService } from "../reasoner/N3Reasoner";
 import { HEIMDALL_WEBSOCKET_PROTOCOL } from "../../server/websocketProtocols";
 import { MetricWriter } from '../../evaluation/MetricWriter';
 import * as HEIMDALL_SETUP from '../../config/heimdall_setup.json';
 import { resolveHeimdallWebSocketUrl } from '../../config/heimdallConfig';
-import { loadSourcePodCredentials } from '../../config/privateCredentials';
+import { SourcePodAccess } from './SourcePodAccess';
 const WebSocketClient = require('websocket').client;
 const websocketConnection = require('websocket').connection;
 const parser = new RSPQLParser();
@@ -37,6 +37,7 @@ export class HeimdallInstantiator {
     private readonly client_id?: string;
     private readonly ready_promise: Promise<void>;
     private readonly sharedStreamRegistry?: SharedStreamRegistry;
+    private readonly sourcePodAccess: SourcePodAccess;
     /**
      * Creates an instance of HeimdallInstantiator.
      * @param {string} query - The RSPQL query.
@@ -48,14 +49,15 @@ export class HeimdallInstantiator {
      * @param {string} rules - The rules for the query.
      * @memberof HeimdallInstantiator
      */
-    public constructor(query: string, from_timestamp: number, to_timestamp: number, logger: any, query_type: string, event_emitter: any, rules: string = '', metric_writer: MetricWriter = new MetricWriter(), client_id?: string, sharedStreamRegistry?: SharedStreamRegistry) {
+    public constructor(query: string, from_timestamp: number, to_timestamp: number, logger: any, query_type: string, event_emitter: any, rules: string = '', metric_writer: MetricWriter = new MetricWriter(), client_id?: string, sharedStreamRegistry?: SharedStreamRegistry, sourcePodAccess: SourcePodAccess = sharedStreamRegistry?.sourcePodAccess || new SourcePodAccess()) {
         this.query = query;
         this.rules = rules;
         this.logger = logger;
         this.event_emitter = event_emitter;
         this.metric_writer = metric_writer;
         this.client_id = client_id;
         this.sharedStreamRegistry = sharedStreamRegistry;
+        this.sourcePodAccess = sourcePodAccess;
         this.hash_string = hash_string_md5(query);
         this.rsp_engine = new RSPEngine(query, {
             // This delay is evaluation configuration for the 4 Hz experiment only;
@@ -94,9 +96,7 @@ export class HeimdallInstantiator {
         if (this.stream_array.length !== 0) {
             if (query_type === 'historical+live') {
                 for (const stream of this.stream_array) {
-                    const session_credentials = this.get_session_credentials(stream);
-                    this.logger.info({ query_hashed }, `stream_credentials_retrieved`);
-                    new DecentralizedFileStreamer(stream, session_credentials, this.from_date, this.to_date, this.rsp_engine, this.query, this.logger);
+                    new DecentralizedFileStreamer(stream, this.sourcePodAccess, this.from_date, this.to_date, this.rsp_engine, this.query, this.logger);
                 }
                 await this.subscribeRStream();
                 return true;
@@ -108,7 +108,6 @@ export class HeimdallInstantiator {
                 // must not precede this query's RStream listener.
                 await this.subscribeRStream();
                 for (const stream of this.stream_array) {
-                    this.logger.info({ query_hashed }, `stream_credentials_retrieved`);
                     if (!this.sharedStreamRegistry) throw new Error('Live processing requires a SharedStreamRegistry');
                     const rdfStream = this.rsp_engine.getStream(stream);
                     if (!rdfStream) throw new Error(`RSP engine has no RDF stream for ${stream}`);
@@ -238,17 +237,4 @@ export class HeimdallInstantiator {
             });
         }
     }
-    /**
-     * Get the session credentials for the Solid Pod.
-     * @param {string} stream_name - The name of the stream (i.e the LDES in LDP of the Solid Pod).
-     * @returns {Credentials} - The session credentials.
-     * @memberof HeimdallInstantiator
-     */
-    get_session_credentials(stream_name: string) {
-        const credentials: Credentials = loadSourcePodCredentials();
-        const session_credentials = credentials[stream_name];
-        if (!session_credentials) throw new Error(`No source-Pod credentials configured for stream ${stream_name}`);
-        return session_credentials;
-    }
-
 }

src/service/heimdall/SharedStreamRegistry.test.ts

@@ -1,5 +1,6 @@
 import { SharedStreamRegistry } from './SharedStreamRegistry';
 import { MetricWriter } from '../../evaluation/MetricWriter';
+import { SourcePodAccess } from './SourcePodAccess';
 import * as fs from 'fs';
 import * as os from 'os';
 import * as path from 'path';
@@ -84,6 +85,41 @@ describe('SharedStreamRegistry', () => {
         expect(registry.consumerCount(streams[0])).toBe(5);
     });
 
+    it('keeps one protected physical acquisition while its source session is reused', async () => {
+        const authenticatedFetch = jest.fn();
+        const createSession = jest.fn().mockResolvedValue({ fetch: authenticatedFetch });
+        const access = new SourcePodAccess({
+            'http://example.test/': { id: 'test-id', secret: 'test-secret', idp: 'https://issuer.example/' },
+        }, createSession);
+        await access.fetchFor(streams[0]);
+        const { dependencies } = makeRegistry();
+        const registry = new SharedStreamRegistry(logger, undefined, dependencies, access);
+        await Promise.all(['Q1', 'Q2'].map((id) => registry.attach(streams[0], id, fakeRdfStream())));
+        expect(dependencies.createSubscription).toHaveBeenCalledTimes(1);
+        expect(createSession).toHaveBeenCalledTimes(1);
+    });
+
+    it('uses one authenticated source fetch for protected live notification discovery and creation', async () => {
+        const authenticatedFetch = jest.fn()
+            .mockResolvedValueOnce({ text: jest.fn().mockResolvedValue(`<http://example.test/A/> <http://www.w3.org/ns/ldp#inbox> <inbox/> .`) })
+            .mockResolvedValueOnce({ headers: new Headers({ link: '<http://example.test/.well-known/solid>; rel="http://www.w3.org/ns/solid/terms#storageDescription"' }) })
+            .mockResolvedValueOnce({ text: jest.fn().mockResolvedValue(`
+                <http://example.test/> <http://www.w3.org/ns/solid/notifications#subscription> <http://example.test/subscribe> .
+                <http://example.test/subscribe> <http://www.w3.org/ns/solid/notifications#channelType> <http://www.w3.org/ns/solid/notifications#WebhookChannel2023> .
+            `) })
+            .mockResolvedValueOnce({ ok: true, text: jest.fn().mockResolvedValue('subscription') });
+        const createSession = jest.fn().mockResolvedValue({ fetch: authenticatedFetch });
+        const access = new SourcePodAccess({
+            'http://example.test/': { id: 'test-id', secret: 'test-secret', idp: 'https://issuer.example/' },
+        }, createSession);
+        const registry = new SharedStreamRegistry(logger, undefined, { readBucketStrategy: jest.fn().mockResolvedValue('http://example.test/time') }, access);
+
+        await registry.attach(streams[0], 'Q1', fakeRdfStream());
+        expect(createSession).toHaveBeenCalledTimes(1);
+        expect(authenticatedFetch).toHaveBeenCalledTimes(4);
+        expect(authenticatedFetch).toHaveBeenNthCalledWith(4, 'http://example.test/subscribe', expect.objectContaining({ method: 'POST' }));
+    });
+
     it('rolls back a failed creation so a later attach retries', async () => {
         const createSubscription = jest.fn().mockRejectedValueOnce(new Error('failed')).mockResolvedValueOnce('subscription');
         const { registry } = makeRegistry({ createSubscription });