merge: cobalt 11 with local processing & better performance (#1287)

Author: wukko

.github/workflows/test.yml

@@ -24,6 +24,8 @@ jobs:
           node-version: 'lts/*'
       - uses: pnpm/action-setup@v4
       - run: .github/test.sh web
+        env:
+          WEB_DEFAULT_API: ${{ vars.WEB_DEFAULT_API }}
 
   test-api:
     name: api sanity check

api/README.md

@@ -71,7 +71,7 @@ as long as you:
 
 ## open source acknowledgements
 ### ffmpeg
-cobalt relies on ffmpeg for muxing and encoding media files. ffmpeg is absolutely spectacular and we're privileged to have an ability to use it for free, just like anyone else. we believe it should be way more recognized.
+cobalt relies on ffmpeg for muxing and encoding media files. ffmpeg is absolutely spectacular and we're privileged to have the ability to use it for free, just like anyone else. we believe it should be way more recognized.
 
 you can [support ffmpeg here](https://ffmpeg.org/donations.html)!
 

api/package.json

@@ -1,7 +1,7 @@
 {
     "name": "@imput/cobalt-api",
     "description": "save what you love",
-    "version": "10.9.4",
+    "version": "11.0",
     "author": "imput",
     "exports": "./src/cobalt.js",
     "type": "module",
@@ -34,6 +34,7 @@
         "ffmpeg-static": "^5.1.0",
         "hls-parser": "^0.10.7",
         "ipaddr.js": "2.2.0",
+        "mime": "^4.0.4",
         "nanoid": "^5.0.9",
         "set-cookie-parser": "2.6.0",
         "undici": "^5.19.1",

api/src/config.js

@@ -1,92 +1,32 @@
-import { Constants } from "youtubei.js";
 import { getVersion } from "@imput/version-info";
-import { services } from "./processing/service-config.js";
-import { supportsReusePort } from "./misc/cluster.js";
+import { loadEnvs, validateEnvs, setupEnvWatcher } from "./core/env.js";
+import * as cluster from "./misc/cluster.js";
 
 const version = await getVersion();
 
-const disabledServices = process.env.DISABLED_SERVICES?.split(',') || [];
-const enabledServices = new Set(Object.keys(services).filter(e => {
-    if (!disabledServices.includes(e)) {
-        return e;
-    }
-}));
-
-const env = {
-    apiURL: process.env.API_URL || '',
-    apiPort: process.env.API_PORT || 9000,
-    tunnelPort: process.env.API_PORT || 9000,
-
-    listenAddress: process.env.API_LISTEN_ADDRESS,
-    freebindCIDR: process.platform === 'linux' && process.env.FREEBIND_CIDR,
-
-    corsWildcard: process.env.CORS_WILDCARD !== '0',
-    corsURL: process.env.CORS_URL,
-
-    cookiePath: process.env.COOKIE_PATH,
-
-    rateLimitWindow: (process.env.RATELIMIT_WINDOW && parseInt(process.env.RATELIMIT_WINDOW)) || 60,
-    rateLimitMax: (process.env.RATELIMIT_MAX && parseInt(process.env.RATELIMIT_MAX)) || 20,
-
-    sessionRateLimitWindow: (process.env.SESSION_RATELIMIT_WINDOW && parseInt(process.env.SESSION_RATELIMIT_WINDOW)) || 60,
-    sessionRateLimit: (process.env.SESSION_RATELIMIT && parseInt(process.env.SESSION_RATELIMIT)) || 10,
-
-    durationLimit: (process.env.DURATION_LIMIT && parseInt(process.env.DURATION_LIMIT)) || 10800,
-    streamLifespan: (process.env.TUNNEL_LIFESPAN && parseInt(process.env.TUNNEL_LIFESPAN)) || 90,
-
-    processingPriority: process.platform !== 'win32'
-        && process.env.PROCESSING_PRIORITY
-        && parseInt(process.env.PROCESSING_PRIORITY),
-
-    externalProxy: process.env.API_EXTERNAL_PROXY,
-
-    turnstileSitekey: process.env.TURNSTILE_SITEKEY,
-    turnstileSecret: process.env.TURNSTILE_SECRET,
-    jwtSecret: process.env.JWT_SECRET,
-    jwtLifetime: process.env.JWT_EXPIRY || 120,
-
-    sessionEnabled: process.env.TURNSTILE_SITEKEY
-                        && process.env.TURNSTILE_SECRET
-                        && process.env.JWT_SECRET,
-
-    apiKeyURL: process.env.API_KEY_URL && new URL(process.env.API_KEY_URL),
-    authRequired: process.env.API_AUTH_REQUIRED === '1',
-    redisURL: process.env.API_REDIS_URL,
-    instanceCount: (process.env.API_INSTANCE_COUNT && parseInt(process.env.API_INSTANCE_COUNT)) || 1,
-    keyReloadInterval: 900,
-
-    enabledServices,
-
-    customInnertubeClient: process.env.CUSTOM_INNERTUBE_CLIENT,
-    ytSessionServer: process.env.YOUTUBE_SESSION_SERVER,
-    ytSessionReloadInterval: 300,
-    ytSessionInnertubeClient: process.env.YOUTUBE_SESSION_INNERTUBE_CLIENT,
-}
+const env = loadEnvs();
 
 const genericUserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36";
 const cobaltUserAgent = `cobalt/${version} (+https://github.com/imputnet/cobalt)`;
 
+export const canonicalEnv = Object.freeze(structuredClone(process.env));
 export const setTunnelPort = (port) => env.tunnelPort = port;
 export const isCluster = env.instanceCount > 1;
+export const updateEnv = (newEnv) => {
+    // tunnelPort is special and needs to get carried over here
+    newEnv.tunnelPort = env.tunnelPort;
 
-if (env.sessionEnabled && env.jwtSecret.length < 16) {
-    throw new Error("JWT_SECRET env is too short (must be at least 16 characters long)");
-}
+    for (const key in env) {
+        env[key] = newEnv[key];
+    }
 
-if (env.instanceCount > 1 && !env.redisURL) {
-    throw new Error("API_REDIS_URL is required when API_INSTANCE_COUNT is >= 2");
-} else if (env.instanceCount > 1 && !await supportsReusePort()) {
-    console.error('API_INSTANCE_COUNT is not supported in your environment. to use this env, your node.js');
-    console.error('version must be >= 23.1.0, and you must be running a recent enough version of linux');
-    console.error('(or other OS that supports it). for more info, see `reusePort` option on');
-    console.error('https://nodejs.org/api/net.html#serverlistenoptions-callback');
-    throw new Error('SO_REUSEPORT is not supported');
+    cluster.broadcast({ env_update: newEnv });
 }
 
-if (env.customInnertubeClient && !Constants.SUPPORTED_CLIENTS.includes(env.customInnertubeClient)) {
-    console.error("CUSTOM_INNERTUBE_CLIENT is invalid. Provided client is not supported.");
-    console.error(`Supported clients are: ${Constants.SUPPORTED_CLIENTS.join(', ')}\n`);
-    throw new Error("Invalid CUSTOM_INNERTUBE_CLIENT");
+await validateEnvs(env);
+
+if (env.envFile) {
+    setupEnvWatcher();
 }
 
 export {

api/src/core/api.js

@@ -8,16 +8,17 @@ import jwt from "../security/jwt.js";
 import stream from "../stream/stream.js";
 import match from "../processing/match.js";
 
-import { env, isCluster, setTunnelPort } from "../config.js";
+import { env } from "../config.js";
 import { extract } from "../processing/url.js";
-import { Green, Bright, Cyan } from "../misc/console-text.js";
+import { Bright, Cyan } from "../misc/console-text.js";
 import { hashHmac } from "../security/secrets.js";
 import { createStore } from "../store/redis-ratelimit.js";
 import { randomizeCiphers } from "../misc/randomize-ciphers.js";
 import { verifyTurnstileToken } from "../security/turnstile.js";
 import { friendlyServiceName } from "../processing/service-alias.js";
-import { verifyStream, getInternalStream } from "../stream/manage.js";
+import { verifyStream } from "../stream/manage.js";
 import { createResponse, normalizeRequest, getIP } from "../processing/request.js";
+import { setupTunnelHandler } from "./itunnel.js";
 
 import * as APIKeys from "../security/api-keys.js";
 import * as Cookies from "../processing/cookie/manager.js";
@@ -47,28 +48,31 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
     const startTime = new Date();
     const startTimestamp = startTime.getTime();
 
-    const serverInfo = JSON.stringify({
-        cobalt: {
-            version: version,
-            url: env.apiURL,
-            startTime: `${startTimestamp}`,
-            durationLimit: env.durationLimit,
-            turnstileSitekey: env.sessionEnabled ? env.turnstileSitekey : undefined,
-            services: [...env.enabledServices].map(e => {
-                return friendlyServiceName(e);
-            }),
-        },
-        git,
-    })
+    const getServerInfo = () => {
+        return JSON.stringify({
+            cobalt: {
+                version: version,
+                url: env.apiURL,
+                startTime: `${startTimestamp}`,
+                turnstileSitekey: env.sessionEnabled ? env.turnstileSitekey : undefined,
+                services: [...env.enabledServices].map(e => {
+                    return friendlyServiceName(e);
+                }),
+            },
+            git,
+        });
+    }
+
+    const serverInfo = getServerInfo();
 
     const handleRateExceeded = (_, res) => {
-        const { status, body } = createResponse("error", {
+        const { body } = createResponse("error", {
             code: "error.api.rate_exceeded",
             context: {
                 limit: env.rateLimitWindow
             }
         });
-        return res.status(status).json(body);
+        return res.status(429).json(body);
     };
 
     const keyGenerator = (req) => hashHmac(getIP(req), 'rate').toString('base64url');
@@ -94,14 +98,14 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
     });
 
     const apiTunnelLimiter = rateLimit({
-        windowMs: env.rateLimitWindow * 1000,
-        limit: (req) => req.rateLimitMax || env.rateLimitMax,
+        windowMs: env.tunnelRateLimitWindow * 1000,
+        limit: env.tunnelRateLimitMax,
         standardHeaders: 'draft-6',
         legacyHeaders: false,
-        keyGenerator: req => req.rateLimitKey || keyGenerator(req),
+        keyGenerator: req => keyGenerator(req),
         store: await createStore('tunnel'),
         handler: (_, res) => {
-            return res.sendStatus(429)
+            return res.sendStatus(429);
         }
     });
 
@@ -180,6 +184,7 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
             }
 
             req.rateLimitKey = hashHmac(token, 'rate');
+            req.isSession = true;
         } catch {
             return fail(res, "error.api.generic");
         }
@@ -244,6 +249,7 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
         if (!parsed) {
             return fail(res, "error.api.link.invalid");
         }
+
         if ("error" in parsed) {
             let context;
             if (parsed?.context) {
@@ -257,13 +263,23 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
                 host: parsed.host,
                 patternMatch: parsed.patternMatch,
                 params: normalizedRequest,
+                isSession: req.isSession ?? false,
             });
 
             res.status(result.status).json(result.body);
         } catch {
             fail(res, "error.api.generic");
         }
-    })
+    });
+
+    app.use('/tunnel', cors({
+        methods: ['GET'],
+        exposedHeaders: [
+            'Estimated-Content-Length',
+            'Content-Disposition'
+        ],
+        ...corsConfig,
+    }));
 
     app.get('/tunnel', apiTunnelLimiter, async (req, res) => {
         const id = String(req.query.id);
@@ -294,35 +310,11 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
         }
 
         return stream(res, streamInfo);
-    })
-
-    const itunnelHandler = (req, res) => {
-        if (!req.ip.endsWith('127.0.0.1')) {
-            return res.sendStatus(403);
-        }
-
-        if (String(req.query.id).length !== 21) {
-            return res.sendStatus(400);
-        }
-
-        const streamInfo = getInternalStream(req.query.id);
-        if (!streamInfo) {
-            return res.sendStatus(404);
-        }
-
-        streamInfo.headers = new Map([
-            ...(streamInfo.headers || []),
-            ...Object.entries(req.headers)
-        ]);
-
-        return stream(res, { type: 'internal', data: streamInfo });
-    };
-
-    app.get('/itunnel', itunnelHandler);
+    });
 
     app.get('/', (_, res) => {
         res.type('json');
-        res.status(200).send(serverInfo);
+        res.status(200).send(env.envFile ? getServerInfo() : serverInfo);
     })
 
     app.get('/favicon.ico', (req, res) => {
@@ -342,10 +334,6 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
     setInterval(randomizeCiphers, 1000 * 60 * 30); // shuffle ciphers every 30 minutes
 
     if (env.externalProxy) {
-        if (env.freebindCIDR) {
-            throw new Error('Freebind is not available when external proxy is enabled')
-        }
-
         setGlobalDispatcher(new ProxyAgent(env.externalProxy))
     }
 
@@ -384,17 +372,5 @@ export const runAPI = async (express, app, __dirname, isPrimary = true) => {
         }
     });
 
-    if (isCluster) {
-        const istreamer = express();
-        istreamer.get('/itunnel', itunnelHandler);
-        const server = istreamer.listen({
-            port: 0,
-            host: '127.0.0.1',
-            exclusive: true
-        }, () => {
-            const { port } = server.address();
-            console.log(`${Green('[✓]')} cobalt sub-instance running on 127.0.0.1:${port}`);
-            setTunnelPort(port);
-        });
-    }
+    setupTunnelHandler();
 }