Generating secrets and storing in config

Questions and Help
#1

I’m trying to generate a secret that stays the same until the lifecycle of the app, however looks like the secret gets overwritten as it gets recreated when starting Sanic. It is similar to this topic, but wanted to provide more context about my use case.

Here is the minimal code:

from sanic import Sanic, text
from secrets import token_hex

app = Sanic("test")

app.config.MY_CUSTOM_SECRET = token_hex(20)

@app.main_process_start
async def start(app, _):
    print("starting with", app.config.MY_CUSTOM_SECRET)

@app.get("/")
async def home(request):
    return text("current " + request.app.config.MY_CUSTOM_SECRET)

if __name__ == "__main__":
    app.run(debug = True, auto_reload = False, workers = 1)

starting with 9176a9e8957288d27dd88351994c091b1f62b892
current 19abe08904a068cc8c8f5626032c89e02b9753ec

My specific use case

I’m trying to launch aria2c in Sanic as subprocess with an RPC secret. I don’t want to keep the secret static. (so saving the secret in an environment variable is not an option for me), I’m using the built-in secrets module to create a random string in each boot.

However, during the startup, the secret I generated doesn’t match the one with the secret I passed to Popen, so I can’t connect with RPC.

import os
from xmlrpc.client import ServerProxy
from subprocess import Popen
from sanic import Sanic, json
from secrets import token_hex

def aria2_worker(secret : str):
    try:
        proc = Popen([
            "aria2c", "--enable-rpc=true", "--rpc-listen-port=7419", 
            ("--rpc-secret=" + secret.removeprefix('token:'))
        ])
        proc.wait(timeout = None)
    except KeyboardInterrupt:
        pass

app = Sanic("example")

@app.main_process_start
async def main_process_start(app):
    # Generate a new secret
    app.ctx.RPC_SECRET = "token:" + token_hex(30)

@app.main_process_ready
async def read(app : Sanic, _):
    # Start aria2 RPC server with worker manager
    app.manager.manage("Aria2", aria2_worker, { "secret": app.ctx.RPC_SECRET })

@app.route("/")
async def aria_info(request : Request):
    client = ServerProxy("http://localhost:7419/rpc")
    return json(client.aria2.tellActive(app.ctx.RPC_SECRET))
    # >>> Fails because RPC secret doesn't match with the secret I passed to aria2_parameters().

Then I came across to shared_ctx, and it keeps the values unchanged which is what I want, but however looks like I’m not expected to use shared_ctx with a str value.

I know that I can just ignore the warning and continue using shared_ctx, but if shared_ctx is the only option for me, then I would prefer not get a warning for not using the normal ctx. Should I continue using shared_ctx or any alternative ways to achieve this?

Sanic version: 22.12.0

Thanks!

#2

shared_ctx is the correct place, consider converting the str to bytes before storage - should eliminate the warning IIRC.

#3

Nope, it didn’t eliminate the warning.

Unsafe object RPC_SECRET with type <class 'bytes'> was added to shared_ctx. 
It may not not function as intended. Consider using the regular ctx.

Looks like Sanic just looks for types under multiprocessing or ctypes.

if not any(
            module.startswith(prefix)
            for prefix in ("multiprocessing", "ctypes")
        ):
            error_logger.warning(
                f"{Colors.YELLOW}Unsafe object {Colors.PURPLE}{name} "
                f"{Colors.YELLOW}with type {Colors.PURPLE}{type(value)} "
                f"{Colors.YELLOW}was added to shared_ctx. It may not "
                "not function as intended. Consider using the regular "
                f"ctx.\nFor more information, please see https://sanic.dev/en"
                "/guide/deployment/manager.html#using-shared-context-between-"
                f"worker-processes.{Colors.END}"
            )
#4

I would just ignore it and use bytes. I am going to add that explicitly in the next release.

1 Like