Python secrets 模块，token_urlsafe() 实例源码

def test_brotli_dynamic_timeout(app):
    from secrets import token_urlsafe

    libforget.brotli.brotli(app, timeout=0.001)

    @app.route('/hard_to_compress')
    def hard_to_compress():  # pylint: disable=W0612
        return token_urlsafe(2**16)

    client = app.test_client()

    resp = client.get(
            '/hard_to_compress',
            headers=[('accept-encoding', 'gzip, br')])

    assert resp.headers.get('x-brotli-cache') == 'TIMEOUT'
    assert resp.headers.get('content-encoding') != 'br'

def make_app():

    settings = {
        "cookie_secret": token_urlsafe(32),
        "login_url": "/login",
        "xsrf_cookies": True,
        "debug": True
    }

    return tornado.web.Application([
        (r"/", MainHandler),
        (r"/static/(.*)", ResourceHandler),
        (r"/fetch", Aor.FetchWebsocket),
        (r"/login", LoginManager),
        (r"/admin", AdminInterface),
        (r"/admin/delete/([0-9]+)", AdminInterfaceDelete),
        (r"/aor", Aor.Aor)
    ], **settings)

def application(app_id=None):
    """Register an application client."""
    form = ApplicationFrom()
    if app_id:
        client = Client.select().where(Client.id == app_id).first()
    else:
        client = Client.select().where(Client.user_id == current_user.id).first()
    if client:
        flash(
            f"You aready have registered application '{client.name}' and issued API credentials.",
            "info")
        return redirect(url_for("api_credentials", app_id=client.id))

    if form.validate_on_submit():
        client = Client(org_id=current_user.organisation.id)
        form.populate_obj(client)
        client.client_id = secrets.token_hex(10)
        client.client_secret = secrets.token_urlsafe(20)
        client.save()
        flash(f"Application '{client.name}' was successfully registered.", "success")
        return redirect(url_for("api_credentials", app_id=client.id))

    return render_template("application.html", form=form)

def token(self):
        if not self.hmac_key:
            t = self.redis.get(self.key('hmac_key'))
            if not t:
                t = secrets.token_urlsafe().encode('ascii')
                self.redis.set(self.key('hmac_key'), t)
            self.hmac_key = t
        return self.hmac_key

def post(self):
        username = self.get_body_argument("username")
        password = self.get_body_argument("password").encode('utf8')
        admin_password = config.Get("admin_password").encode('utf8')
        if username == config.Get("admin_username") and bcrypt.checkpw(password, admin_password):
            self.set_secure_cookie("i", token_urlsafe(32))
            self.redirect("/admin")
        else:
            self.redirect("/login")

def _get_state(self):
        return secrets.token_urlsafe(16)

def test_file_upload(session, tmpdir):
    path = Path(str(tmpdir)) / 'file.txt'
    payload = secrets.token_urlsafe()
    with path.open('w') as fobj:
        fobj.write(payload)
    await session.get('/file/')
    file_input = await session.wait_for_element(5, 'input[name="file"]')
    await file_input.send_file(path)
    submit_input = await session.get_element('input[type="submit"]')
    await submit_input.click()
    contents_span = await session.wait_for_element(5, '#contents')
    assert payload == await contents_span.get_text()

def register(user_id: int) -> Token:
    """Generate a new token for a user"""

    token = secrets.token_urlsafe(token_length)

    with await db.cachepool as cache:
        await cache.set(f"users:{user_id}:tokens:{token}", 1)
        await cache.expire(f"users:{user_id}:tokens:{token}", token_validity)

    return token

def create_for(guest: db.Guest) -> Challenge:
    """Create a challenge for a db.Guest"""

    chlg = token_urlsafe(challenge_length)
    with await db.cachepool as cache:
        await cache.hmset_dict(f"chlgs:{chlg}", guest._asdict())
        await cache.expire(f"chlgs:{chlg}", challenge_validity)
    return chlg

def utility_processor():  # noqa: D202
    """Define funcions callable form Jinja2 using application context."""

    def onboarded_organisations():
        return list(
            Organisation.select(Organisation.name, Organisation.tuakiri_name).where(
                Organisation.confirmed.__eq__(True)))

    def orcid_login_url():
        return url_for("orcid_login", next=get_next_url())

    def tuakiri_login_url():
        _next = get_next_url()
        if EXTERNAL_SP:
            session["auth_secret"] = secret_token = secrets.token_urlsafe()
            _next = url_for("handle_login", _next=_next, _external=True)
            login_url = append_qs(EXTERNAL_SP, _next=_next, key=secret_token)
        else:
            login_url = url_for("handle_login", _next=_next)
        return login_url

    return dict(
        orcid_login_url=orcid_login_url,
        tuakiri_login_url=tuakiri_login_url,
        onboarded_organisations=onboarded_organisations,
    )

def api_credentials(app_id=None):
    """Manage API credentials."""
    if app_id:
        client = Client.select().where(Client.id == app_id).first()
    else:
        client = Client.select().where(Client.user_id == current_user.id).first()
    if not client:
        return redirect(url_for("application"))
    form = CredentialForm(obj=client)
    print(form.reset.data)
    print("***", request.form)
    if form.validate_on_submit():
        if form.revoke.data:
            Token.delete().where(Token.client == client).execute()
        elif form.reset.data:
            form.client_id.data = client.client_id = secrets.token_hex(10)
            form.client_secret.data = client.client_secret = secrets.token_urlsafe(20)
            client.save()
        elif form.update_app.data:
            form.populate_obj(client)
            client.save()
        elif form.delete.data:
            Token.delete().where(Token.client == client).execute()
            client.delete().execute()
            return redirect(url_for("application"))

    return render_template("api_credentials.html", form=form)

def token_urlsafe(nbytes=32):
        token = base64.urlsafe_b64encode(token_bytes(nbytes)).rstrip(b'=')
        return token.decode('ascii') if PY3 else token

def apply_short_links(context, click_url, click_random=30, backup_arg=False):
    shortened_link = []
    extra = {}
    for k, v in context.items():
        # TODO deal with unsubscribe links properly
        if k != 'unsubscribe_link' and looks_like_link(v):
            r = secrets.token_urlsafe(click_random)[:click_random]
            new_url = click_url + r
            if backup_arg:
                new_url += '?u=' + urlsafe_b64encode(v.encode()).decode()
            extra[k] = new_url
            extra[f'{k}_original'] = v
            shortened_link.append((v, r))
    context.update(extra)
    return shortened_link

def tokhex(length=10, urlsafe=False):
    import secrets
    if urlsafe == True:
        return secrets.token_urlsafe(length)
    else:
        return secrets.token_hex(length)

# Show A Type Of Face

def convert(self, data):
        # data arg is interpreted as the desired count of characters to generate
        if not isinstance(data, int):
            try:
                data = int(data, base=10)
            except:
                data = 8
        if not data:
            return ()

        # value returned by secrets.token_urlsafe is base64-encoded so longer
        # than requested
        return (secrets.token_urlsafe(data)[0:data], )