1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
# Copyright (C) VyOS Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import datetime
import threading
from secrets import token_hex
import jwt
from ...session import SessionState
_secret_lock = threading.Lock()
def verify_oidc_token(token: str):
"""Validate an OIDC Bearer token against the configured OIDC issuer JWKS.
Returns the subject identifier on success, None on failure.
Requires set service https api rest authentication oidc issuer <url>.
"""
state = SessionState()
if not state.oidc_issuer:
return None
try:
from jwt import PyJWKClient
# Use explicit jwks-url if set, otherwise use OIDC discovery
if state.oidc_jwks_uri:
jwks_uri = state.oidc_jwks_uri
else:
import urllib.request as _req
import json as _json
disc_url = f"{state.oidc_issuer}/.well-known/openid-configuration"
with _req.urlopen(disc_url, timeout=10) as r:
jwks_uri = _json.loads(r.read())["jwks_uri"]
state.oidc_jwks_uri = jwks_uri
jwks_client = PyJWKClient(jwks_uri, cache_keys=True)
signing_key = jwks_client.get_signing_key_from_jwt(token)
decode_options = {"require": ["exp"]}
decode_kwargs: dict = dict(
algorithms=["RS256", "RS384", "RS512", "ES256", "ES384", "ES512"],
options=decode_options,
issuer=state.oidc_issuer,
leeway=30,
)
if not state.oidc_audience:
# Fail closed: config-time validation in service_https.py's
# verify() requires oidc.audience whenever oidc.issuer is
# set, but never trust that alone for a security-critical
# check. PyJWT also skips aud validation whenever no
# "audience" kwarg is passed at all, independent of
# verify_aud -- so this must refuse outright, not just stop
# forcing verify_aud=False.
return None
decode_kwargs["audience"] = state.oidc_audience
payload = jwt.decode(token, signing_key.key, **decode_kwargs)
return payload.get("sub") or payload.get("client_id")
except Exception:
return None
def init_secret():
state = SessionState()
if state.rest_secret is not None:
return
length = int(state.rest_secret_len or 32)
with _secret_lock:
if state.rest_secret is None:
state.rest_secret = token_hex(length)
def generate_token(key_id: str) -> dict:
state = SessionState()
init_secret()
exp_interval = int(state.rest_token_exp or 3600)
expiration = datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(
seconds=exp_interval
)
payload = {'iss': 'vyos-rest-api', 'sub': key_id, 'exp': expiration}
token = jwt.encode(payload=payload, key=state.rest_secret, algorithm='HS256')
return {'token': token, 'expires_in': exp_interval}
def verify_token(token: str):
state = SessionState()
if state.rest_secret is None:
return None
try:
payload = jwt.decode(token, state.rest_secret, algorithms=['HS256'])
except jwt.ExpiredSignatureError:
return None
except jwt.PyJWTError:
return None
return payload.get('sub')
|
