Handle upstream policies too

src/api/controllers/TermsController.ts

@@ -5,6 +5,12 @@ import TermsTextRecord from "../../db/models/TermsTextRecord";
 import TermsSignedRecord from "../../db/models/TermsSignedRecord";
 import { Op } from "sequelize";
 import { Cache, CACHE_TERMS } from "../../MemoryCache";
+import UserScalarToken from "../../db/models/UserScalarToken";
+import Upstream from "../../db/models/Upstream";
+import { LogService } from "matrix-js-snippets";
+import { ScalarClient } from "../../scalar/ScalarClient";
+import { md5 } from "../../utils/hashing";
+import TermsUpstreamRecord from "../../db/models/TermsUpstreamRecord";
 
 export interface ILanguagePolicy {
     name: string;
@@ -85,8 +91,6 @@ export default class TermsController {
     }
 
     public async getMissingTermsForUser(user: IMSCUser): Promise<ITermsNotSignedResponse> {
-        // TODO: Upstream policies
-
         const latest = await this.getPublishedTerms();
         const signed = await TermsSignedRecord.findAll({where: {userId: user.userId}});
 
@@ -106,6 +110,47 @@ export default class TermsController {
             }
         }
 
+        // Get upstream terms for the user
+        const tokensForUser = await UserScalarToken.findAll({where: {userId: user.userId}, include: [Upstream]});
+        const upstreamTokens = tokensForUser.filter(t => t.upstream);
+        const urlsToUpstream = {}; // {url: [upstreamId]}
+        for (const upstreamToken of upstreamTokens) {
+            try {
+                const scalarClient = new ScalarClient(upstreamToken.upstream, ScalarClient.KIND_MATRIX_V1);
+                const upstreamTerms = await scalarClient.getMissingTerms(upstreamToken.scalarToken);
+
+                // rewrite the shortcodes to avoid conflicts
+                const shortcodePrefix = `upstream_${md5(`${upstreamToken.id}:${upstreamToken.upstream.apiUrl}`)}`;
+                for (const shortcode of Object.keys(upstreamTerms.policies)) {
+                    policies.policies[`${shortcodePrefix}_${shortcode}`] = upstreamTerms.policies[shortcode];
+
+                    // copy all urls for later adding to the database
+                    for (const language of Object.keys(upstreamTerms.policies[shortcode])) {
+                        const upstreamUrl = upstreamTerms.policies[shortcode][language]['url'];
+                        if (!urlsToUpstream[upstreamUrl]) urlsToUpstream[upstreamUrl] = [];
+                        const upstreamsArr = urlsToUpstream[upstreamUrl];
+                        if (!upstreamsArr.includes(upstreamToken.upstream.id)) {
+                            upstreamsArr.push(upstreamToken.upstream.id);
+                        }
+                    }
+                }
+            } catch (e) {
+                LogService.error("TermsController", e);
+            }
+        }
+
+        // actually cache the urls in the database
+        const existingCache = await TermsUpstreamRecord.findAll({where: {url: {[Op.in]: Object.keys(urlsToUpstream)}}});
+        for (const upstreamUrl of Object.keys(urlsToUpstream)) {
+            const upstreamIds = urlsToUpstream[upstreamUrl];
+            const existingIds = existingCache.filter(c => c.url === upstreamUrl).map(c => c.upstreamId);
+            const missingIds = upstreamIds.filter(i => !existingIds.includes(i));
+            for (const targetUpstreamId of missingIds) {
+                const item = await TermsUpstreamRecord.create({url: upstreamUrl, upstreamId: targetUpstreamId});
+                existingCache.push(item);
+            }
+        }
+
         return policies;
     }
 
@@ -117,6 +162,37 @@ export default class TermsController {
         for (const termsToSign of toAdd) {
             await TermsSignedRecord.create({termsId: termsToSign.id, userId: user.userId});
         }
+
+        // Check upstreams too, if there are any
+        const upstreamPolicies = await TermsUpstreamRecord.findAll({
+            where: {url: {[Op.in]: urls}},
+            include: [Upstream]
+        });
+        const upstreamsToSignatures: { [upstreamId: number]: { upstream: Upstream, token: string, urls: string[] } } = {};
+        for (const upstreamPolicy of upstreamPolicies) {
+            const userToken = await UserScalarToken.findOne({
+                where: {
+                    upstreamId: upstreamPolicy.upstreamId,
+                    userId: user.userId,
+                },
+            });
+            if (!userToken) {
+                LogService.warn("TermsController", `User ${user.userId} is missing an upstream token for ${upstreamPolicy.upstream.scalarUrl}`);
+                continue;
+            }
+
+            if (!upstreamsToSignatures[upstreamPolicy.upstreamId]) upstreamsToSignatures[upstreamPolicy.upstreamId] = {
+                upstream: upstreamPolicy.upstream,
+                token: userToken.scalarToken,
+                urls: [],
+            };
+            upstreamsToSignatures[upstreamPolicy.upstreamId].urls.push(upstreamPolicy.url);
+        }
+
+        for (const upstreamSignature of Object.values(upstreamsToSignatures)) {
+            const client = new ScalarClient(upstreamSignature.upstream, ScalarClient.KIND_MATRIX_V1);
+            await client.signTermsUrls(upstreamSignature.token, upstreamSignature.urls);
+        }
     }
 
     public async getPoliciesForAdmin(): Promise<ITerms[]> {

src/db/DimensionStore.ts

@@ -29,6 +29,7 @@ import SlackBridgeRecord from "./models/SlackBridgeRecord";
 import TermsRecord from "./models/TermsRecord";
 import TermsTextRecord from "./models/TermsTextRecord";
 import TermsSignedRecord from "./models/TermsSignedRecord";
+import TermsUpstreamRecord from "./models/TermsUpstreamRecord";
 
 class _DimensionStore {
     private sequelize: Sequelize;
@@ -69,6 +70,7 @@ class _DimensionStore {
             TermsRecord,
             TermsTextRecord,
             TermsSignedRecord,
+            TermsUpstreamRecord,
         ]);
     }
 

src/db/migrations/20190710213945-AddUpstreamTermsCache.ts

@@ -0,0 +1,21 @@
+import { QueryInterface } from "sequelize";
+import { DataType } from "sequelize-typescript";
+
+export default {
+    up: (queryInterface: QueryInterface) => {
+        return Promise.resolve()
+            .then(() => queryInterface.createTable("dimension_terms_upstream", {
+                "id": {type: DataType.INTEGER, primaryKey: true, autoIncrement: true, allowNull: false},
+                "upstreamId": {
+                    type: DataType.INTEGER, allowNull: false,
+                    references: {model: "dimension_upstreams", key: "id"},
+                    onUpdate: "cascade", onDelete: "cascade",
+                },
+                "url": {type: DataType.STRING, allowNull: false},
+            }));
+    },
+    down: (queryInterface: QueryInterface) => {
+        return Promise.resolve()
+            .then(() => queryInterface.dropTable("dimension_terms_upstream"));
+    }
+}

src/db/models/TermsUpstreamRecord.ts

@@ -0,0 +1,34 @@
+import {
+    AllowNull,
+    AutoIncrement,
+    BelongsTo,
+    Column,
+    ForeignKey,
+    Model,
+    PrimaryKey,
+    Table
+} from "sequelize-typescript";
+import Upstream from "./Upstream";
+
+@Table({
+    tableName: "dimension_terms_upstream",
+    underscored: false,
+    timestamps: false,
+})
+export default class TermsUpstreamRecord extends Model<TermsUpstreamRecord> {
+    @PrimaryKey
+    @AutoIncrement
+    @Column
+    id: number;
+
+    @AllowNull
+    @Column
+    @ForeignKey(() => Upstream)
+    upstreamId?: number;
+
+    @BelongsTo(() => Upstream)
+    upstream: Upstream;
+
+    @Column
+    url: string;
+}

src/scalar/ScalarClient.ts

@@ -4,10 +4,12 @@ import * as request from "request";
 import { LogService } from "matrix-js-snippets";
 import Upstream from "../db/models/Upstream";
 import { SCALAR_API_VERSION } from "../utils/common-constants";
+import { ITermsNotSignedResponse } from "../api/controllers/TermsController";
 
 const REGISTER_ROUTE = "/register";
 const ACCOUNT_INFO_ROUTE = "/account";
 const LOGOUT_ROUTE = "/logout";
+const TERMS_ROUTE = "/terms";
 
 export class ScalarClient {
     public static readonly KIND_LEGACY = "legacy";
@@ -30,7 +32,11 @@ export class ScalarClient {
             };
         } else {
             const parsed = new URL(this.upstream.scalarUrl);
-            parsed.pathname = '/_matrix/integrations/v1' + (path === ACCOUNT_INFO_ROUTE ? path : `${ACCOUNT_INFO_ROUTE}${path}`);
+            if (path === ACCOUNT_INFO_ROUTE || path === TERMS_ROUTE) {
+                parsed.pathname = `/_matrix/integrations/v1${path}`;
+            } else {
+                parsed.pathname = `/_matrix/integrations/v1${ACCOUNT_INFO_ROUTE}${path}`;
+            }
 
             const headers = {};
             if (token) headers['Authorization'] = `Bearer ${token}`;
@@ -105,7 +111,7 @@ export class ScalarClient {
                 json: true,
             }, (err, res, _body) => {
                 if (err) {
-                    LogService.error("ScalarClient", "Error getting information for token");
+                    LogService.error("ScalarClient", "Error logging out token");
                     LogService.error("ScalarClient", err);
                     reject(err);
                 } else if (res.statusCode !== 200) {
@@ -117,4 +123,54 @@ export class ScalarClient {
             });
         });
     }
+
+    public getMissingTerms(token: string): Promise<ITermsNotSignedResponse> {
+        const {scalarUrl, headers, queryString} = this.makeRequestArguments(TERMS_ROUTE, token);
+        LogService.info("ScalarClient", "Doing upstream scalar request: GET " + scalarUrl);
+        return new Promise((resolve, reject) => {
+            request({
+                method: "GET",
+                url: scalarUrl,
+                qs: queryString,
+                headers: headers,
+                json: true,
+            }, (err, res, _body) => {
+                if (err) {
+                    LogService.error("ScalarClient", "Error getting terms for token");
+                    LogService.error("ScalarClient", err);
+                    reject(err);
+                } else if (res.statusCode !== 200) {
+                    LogService.error("ScalarClient", "Got status code " + res.statusCode + " while getting terms for token");
+                    reject(res.statusCode);
+                } else {
+                    resolve(res.body);
+                }
+            });
+        });
+    }
+
+    public signTermsUrls(token: string, urls: string[]): Promise<any> {
+        const {scalarUrl, headers, queryString} = this.makeRequestArguments(TERMS_ROUTE, token);
+        LogService.info("ScalarClient", "Doing upstream scalar request: POST " + scalarUrl);
+        return new Promise((resolve, reject) => {
+            request({
+                method: "POST",
+                url: scalarUrl,
+                qs: queryString,
+                headers: headers,
+                json: {user_accepts: urls},
+            }, (err, res, _body) => {
+                if (err) {
+                    LogService.error("ScalarClient", "Error updating terms for token");
+                    LogService.error("ScalarClient", err);
+                    reject(err);
+                } else if (res.statusCode !== 200) {
+                    LogService.error("ScalarClient", "Got status code " + res.statusCode + " while updating terms for token");
+                    reject(res.statusCode);
+                } else {
+                    resolve(res.body);
+                }
+            });
+        });
+    }
 }

src/utils/hashing.ts

@@ -0,0 +1,5 @@
+import * as crypto from "crypto";
+
+export function md5(text: string): string {
+    return crypto.createHash("md5").update(text).digest('hex').toLowerCase();
+}