Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/oauthlib/common.py: 29%

1"""

2oauthlib.common

3~~~~~~~~~~~~~~

5This module provides data structures and utilities common

6to all implementations of OAuth.

7"""

8import collections

9import datetime

10import logging

11import re

12import time

13import urllib.parse as urlparse

14from urllib.parse import (

15 quote as _quote, unquote as _unquote, urlencode as _urlencode,

16)

18from . import get_debug

20try:

21 from secrets import SystemRandom, randbits

22except ImportError:

23 from random import SystemRandom, getrandbits as randbits

25UNICODE_ASCII_CHARACTER_SET = ('abcdefghijklmnopqrstuvwxyz'

26 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'

27 '0123456789')

29CLIENT_ID_CHARACTER_SET = (r' !"#$%&\'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMN'

30 'OPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}')

32SANITIZE_PATTERN = re.compile(r'([^&;]*(?:password|token)[^=]*=)[^&;]+', re.IGNORECASE)

33INVALID_HEX_PATTERN = re.compile(r'%[^0-9A-Fa-f]|%[0-9A-Fa-f][^0-9A-Fa-f]')

35always_safe = ('ABCDEFGHIJKLMNOPQRSTUVWXYZ'

36 'abcdefghijklmnopqrstuvwxyz'

37 '0123456789_.-')

39log = logging.getLogger('oauthlib')

42# 'safe' must be bytes (Python 2.6 requires bytes, other versions allow either)

43def quote(s, safe=b'/'):

44 s = s.encode('utf-8') if isinstance(s, str) else s

45 s = _quote(s, safe)

46 # PY3 always returns unicode. PY2 may return either, depending on whether

47 # it had to modify the string.

48 if isinstance(s, bytes):

49 s = s.decode('utf-8')

50 return s

53def unquote(s):

54 s = _unquote(s)

55 # PY3 always returns unicode. PY2 seems to always return what you give it,

56 # which differs from quote's behavior. Just to be safe, make sure it is

57 # unicode before we return.

58 if isinstance(s, bytes):

59 s = s.decode('utf-8')

60 return s

63def urlencode(params):

64 utf8_params = encode_params_utf8(params)

65 urlencoded = _urlencode(utf8_params)

66 if isinstance(urlencoded, str):

67 return urlencoded

68 else:

69 return urlencoded.decode("utf-8")

72def encode_params_utf8(params):

73 """Ensures that all parameters in a list of 2-element tuples are encoded to

74 bytestrings using UTF-8

75 """

76 encoded = []

77 for k, v in params:

78 encoded.append((

79 k.encode('utf-8') if isinstance(k, str) else k,

80 v.encode('utf-8') if isinstance(v, str) else v))

81 return encoded

84def decode_params_utf8(params):

85 """Ensures that all parameters in a list of 2-element tuples are decoded to

86 unicode using UTF-8.

87 """

88 decoded = []

89 for k, v in params:

90 decoded.append((

91 k.decode('utf-8') if isinstance(k, bytes) else k,

92 v.decode('utf-8') if isinstance(v, bytes) else v))

93 return decoded

96urlencoded = set(always_safe) | set('=&;:%+~,*@!()/?\'$')

99def urldecode(query):

100 """Decode a query string in x-www-form-urlencoded format into a sequence

101 of two-element tuples.

102

103 Unlike urlparse.parse_qsl(..., strict_parsing=True) urldecode will enforce

104 correct formatting of the query string by validation. If validation fails

105 a ValueError will be raised. urllib.parse_qsl will only raise errors if

106 any of name-value pairs omits the equals sign.

107 """

108 # Check if query contains invalid characters

109 if query and not set(query) <= urlencoded:

110 error = ("Error trying to decode a non urlencoded string. "

111 "Found invalid characters: %s "

112 "in the string: '%s'. "

113 "Please ensure the request/response body is "

114 "x-www-form-urlencoded.")

115 raise ValueError(error % (set(query) - urlencoded, query))

116

117 # Check for correctly hex encoded values using a regular expression

118 # All encoded values begin with % followed by two hex characters

119 # correct = %00, %A0, %0A, %FF

120 # invalid = %G0, %5H, %PO

121 if INVALID_HEX_PATTERN.search(query):

122 raise ValueError('Invalid hex encoding in query string.')

123

124 # We want to allow queries such as "c2" whereas urlparse.parse_qsl

125 # with the strict_parsing flag will not.

126 params = urlparse.parse_qsl(query, keep_blank_values=True)

127

128 # unicode all the things

129 return decode_params_utf8(params)

130

131

132def extract_params(raw):

133 """Extract parameters and return them as a list of 2-tuples.

134

135 Will successfully extract parameters from urlencoded query strings,

136 dicts, or lists of 2-tuples. Empty strings/dicts/lists will return an

137 empty list of parameters. Any other input will result in a return

138 value of None.

139 """

140 if isinstance(raw, (bytes, str)):

141 try:

142 params = urldecode(raw)

143 except ValueError:

144 params = None

145 elif hasattr(raw, '__iter__'):

146 try:

147 dict(raw)

148 except ValueError:

149 params = None

150 except TypeError:

151 params = None

152 else:

153 params = list(raw.items() if isinstance(raw, dict) else raw)

154 params = decode_params_utf8(params)

155 else:

156 params = None

157

158 return params

159

160

161def generate_nonce():

162 """Generate pseudorandom nonce that is unlikely to repeat.

163

164 Per `section 3.3`_ of the OAuth 1 RFC 5849 spec.

165 Per `section 3.2.1`_ of the MAC Access Authentication spec.

166

167 RFC 5849 section 3.3 requires the nonce to be unique "across all requests

168 with the same timestamp, client credentials, and token combinations". The

169 ``oauth_timestamp`` parameter is sent separately, so embedding the epoch

170 time in the nonce does nothing to distinguish requests that share a

171 timestamp: within a given second the collision resistance comes entirely

172 from the random component. This uses 96 bits drawn from the system CSPRNG

173 (up from the previous 64 bits appended to the timestamp), which keeps the

174 birthday-bound collision probability negligible while the decimal encoding

175 stays within the default nonce length limits.

176

177 .. _`section 3.2.1`: https://tools.ietf.org/html/draft-ietf-oauth-v2-http-mac-01#section-3.2.1

178 .. _`section 3.3`: https://tools.ietf.org/html/rfc5849#section-3.3

179 """

180 return str(randbits(96))

181

182

183def generate_timestamp():

184 """Get seconds since epoch (UTC).

185

186 Per `section 3.3`_ of the OAuth 1 RFC 5849 spec.

187 Per `section 3.2.1`_ of the MAC Access Authentication spec.

188

189 .. _`section 3.2.1`: https://tools.ietf.org/html/draft-ietf-oauth-v2-http-mac-01#section-3.2.1

190 .. _`section 3.3`: https://tools.ietf.org/html/rfc5849#section-3.3

191 """

192 return str(int(time.time()))

193

194

195def generate_token(length=30, chars=UNICODE_ASCII_CHARACTER_SET):

196 """Generates a non-guessable OAuth token

197

198 OAuth (1 and 2) does not specify the format of tokens except that they

199 should be strings of random characters. Tokens should not be guessable

200 and entropy when generating the random characters is important. Which is

201 why SystemRandom is used instead of the default random.choice method.

202 """

203 rand = SystemRandom()

204 return ''.join(rand.choice(chars) for x in range(length))

205

206

207def generate_signed_token(private_pem, request):

208 import jwt # noqa: PLC0415

209

210 now = datetime.datetime.utcnow()

211

212 claims = {

213 'scope': request.scope,

214 'exp': now + datetime.timedelta(seconds=request.expires_in)

215 }

216

217 claims.update(request.claims)

218

219 token = jwt.encode(claims, private_pem, 'RS256')

220 token = to_unicode(token, "UTF-8")

221

222 return token

223

224

225def verify_signed_token(public_pem, token):

226 import jwt # noqa: PLC0415

227

228 return jwt.decode(token, public_pem, algorithms=['RS256'])

229

230

231def generate_client_id(length=30, chars=CLIENT_ID_CHARACTER_SET):

232 """Generates an OAuth client_id

233

234 OAuth 2 specify the format of client_id in

235 https://tools.ietf.org/html/rfc6749#appendix-A.

236 """

237 return generate_token(length, chars)

238

239

240def add_params_to_qs(query, params):

241 """Extend a query with a list of two-tuples."""

242 if isinstance(params, dict):

243 params = params.items()

244 queryparams = urlparse.parse_qsl(query, keep_blank_values=True)

245 queryparams.extend(params)

246 return urlencode(queryparams)

247

248

249def add_params_to_uri(uri, params, fragment=False):

250 """Add a list of two-tuples to the uri query components."""

251 sch, net, path, par, query, fra = urlparse.urlparse(uri)

252 if fragment:

253 fra = add_params_to_qs(fra, params)

254 else:

255 query = add_params_to_qs(query, params)

256 return urlparse.urlunparse((sch, net, path, par, query, fra))

257

258

259def safe_string_equals(a, b):

260 """ Near-constant time string comparison.

261

262 Used in order to avoid timing attacks on sensitive information such

263 as secret keys during request verification (`rootLabs`_).

264

265 .. _`rootLabs`: http://rdist.root.org/2010/01/07/timing-independent-array-comparison/

266

267 """

268 if len(a) != len(b):

269 return False

270

271 result = 0

272 for x, y in zip(a, b):

273 result |= ord(x) ^ ord(y)

274 return result == 0

275

276

277def check_http_header_value(name, value):

278 """Validate that a value is safe to embed in an HTTP header field value.

279

280 Prevents header injection by rejecting characters that allow the value

281 to break out of its header field, per RFC 9110 section 5.5:

282

283 - CR and LF characters, which enable header splitting/injection;

284 - other control characters (CTLs, including NUL), which may corrupt

285 or terminate parsing of the header by downstream intermediaries;

286 - double quotes, which allow a quoted-string parameter in a header

287 value (such as the MAC Authorization header) to break out and

288 inject arbitrary parameters;

289 - non-ASCII characters, which are mangled inconsistently by proxies

290 and gateways and may be re-interpreted as control characters.

291

292 :param name: Name of the parameter, used in error messages.

293 :param value: Candidate header value. ``None`` is allowed and ignored.

294 :raises ValueError: if the value contains unsafe characters.

295 """

296 if value is None:

297 return

298 if not isinstance(value, str):

299 raise ValueError('%s must be a string' % name)

300 for char in value:

301 if char in '\r\n':

302 raise ValueError('%s must not contain CR or LF characters' % name)

303 if ord(char) < 0x20 or ord(char) == 0x7f:

304 raise ValueError('%s must not contain control characters' % name)

305 if char == '"':

306 raise ValueError('%s must not contain double quotes' % name)

307 if ord(char) > 0x7e:

308 raise ValueError('%s must contain ASCII characters only' % name)

309

310

311def to_unicode(data, encoding='UTF-8'):

312 """Convert a number of different types of objects to unicode."""

313 if isinstance(data, str):

314 return data

315

316 if isinstance(data, bytes):

317 return str(data, encoding=encoding)

318

319 if hasattr(data, '__iter__'):

320 try:

321 dict(data)

322 except TypeError:

323 pass

324 except ValueError:

325 # Assume it's a one dimensional data structure

326 return (to_unicode(i, encoding) for i in data)

327 else:

328 # We support 2.6 which lacks dict comprehensions

329 if hasattr(data, 'items'):

330 data = data.items()

331 return {to_unicode(k, encoding): to_unicode(v, encoding) for k, v in data}

332

333 return data

334

335

336class CaseInsensitiveDict(dict):

337

338 """Basic case insensitive dict with strings only keys."""

339

340 proxy = {}

341

342 def __init__(self, data):

343 self.proxy = {k.lower(): k for k in data}

344 for k in data:

345 self[k] = data[k]

346

347 def __contains__(self, k):

348 return k.lower() in self.proxy

349

350 def __delitem__(self, k):

351 key = self.proxy[k.lower()]

352 super().__delitem__(key)

353 del self.proxy[k.lower()]

354

355 def __getitem__(self, k):

356 key = self.proxy[k.lower()]

357 return super().__getitem__(key)

358

359 def get(self, k, default=None):

360 return self[k] if k in self else default # noqa: SIM401

361

362 def __setitem__(self, k, v):

363 super().__setitem__(k, v)

364 self.proxy[k.lower()] = k

365

366 def update(self, *args, **kwargs):

367 super().update(*args, **kwargs)

368 for k in dict(*args, **kwargs):

369 self.proxy[k.lower()] = k

370

371

372class Request:

373

374 """A malleable representation of a signable HTTP request.

375

376 Body argument may contain any data, but parameters will only be decoded if

377 they are one of:

378

379 * urlencoded query string

380 * dict

381 * list of 2-tuples

382

383 Anything else will be treated as raw body data to be passed through

384 unmolested.

385 """

386

387 def __init__(self, uri, http_method='GET', body=None, headers=None,

388 encoding='utf-8'):

389 # Convert to unicode using encoding if given, else assume unicode

390 def encode(x):

391 return to_unicode(x, encoding) if encoding else x

392

393 self.uri = encode(uri)

394 self.http_method = encode(http_method)

395 self.headers = CaseInsensitiveDict(encode(headers or {}))

396 self.body = encode(body)

397 self.decoded_body = extract_params(self.body)

398 self.oauth_params = []

399 self.validator_log = {}

400

401 self._params = {

402 "access_token": None,

403 "client": None,

404 "client_id": None,

405 "client_secret": None,

406 "code": None,

407 "code_challenge": None,

408 "code_challenge_method": None,

409 "code_verifier": None,

410 "device_code": None,

411 "extra_credentials": None,

412 "grant_type": None,

413 "redirect_uri": None,

414 "refresh_token": None,

415 "request_token": None,

416 "resource": None,

417 "response_type": None,

418 "scope": None,

419 "scopes": None,

420 "state": None,

421 "token": None,

422 "user": None,

423 "token_type_hint": None,

424

425 # OpenID Connect

426 "response_mode": None,

427 "nonce": None,

428 "display": None,

429 "prompt": None,

430 "claims": None,

431 "max_age": None,

432 "ui_locales": None,

433 "id_token_hint": None,

434 "login_hint": None,

435 "acr_values": None

436 }

437 self._params.update(dict(urldecode(self.uri_query)))

438 self._params.update(dict(self.decoded_body or []))

439

440 def __getattr__(self, name):

441 if name in self._params:

442 return self._params[name]

443 else:

444 raise AttributeError(name)

445

446 def __repr__(self):

447 if not get_debug():

448 return "<oauthlib.Request SANITIZED>"

449 body = self.body

450 headers = self.headers.copy()

451 if body:

452 body = SANITIZE_PATTERN.sub(r'\1<SANITIZED>', str(body))

453 if 'Authorization' in headers:

454 headers['Authorization'] = '<SANITIZED>'

455 return '<oauthlib.Request url="{}", http_method="{}", headers="{}", body="{}">'.format(

456 self.uri, self.http_method, headers, body)

457

458 @property

459 def uri_query(self):

460 return urlparse.urlparse(self.uri).query

461

462 @property

463 def uri_query_params(self):

464 if not self.uri_query:

465 return []

466 return urlparse.parse_qsl(self.uri_query, keep_blank_values=True,

467 strict_parsing=True)

468

469 @property

470 def duplicate_params(self):

471 seen_keys = collections.defaultdict(int)

472 all_keys = (p[0]

473 for p in (self.decoded_body or []) + self.uri_query_params)

474 for k in all_keys:

475 seen_keys[k] += 1

476 return [k for k, c in seen_keys.items() if c > 1]