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
|
#---------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
#---------------------------------------------------------------------------------------------
from threading import Lock
try:
import urllib.parse as parse
except ImportError:
import urlparse as parse # pylint: disable=import-error
_cache = {}
_lock = Lock()
def get_challenge_for_url(url):
""" Gets the challenge for the cached URL.
:param url: the URL the challenge is cached for.
:rtype: HttpBearerChallenge """
if not url:
raise ValueError('URL cannot be None')
url = parse.urlparse(url)
_lock.acquire()
val = _cache.get(url.netloc)
_lock.release()
return val
def remove_challenge_for_url(url):
""" Removes the cached challenge for the specified URL.
:param url: the URL for which to remove the cached challenge """
if not url:
raise ValueError('URL cannot be empty')
url = parse.urlparse(url)
_lock.acquire()
del _cache[url.netloc]
_lock.release()
def set_challenge_for_url(url, challenge):
""" Caches the challenge for the specified URL.
:param url: the URL for which to cache the challenge
:param challenge: the challenge to cache """
if not url:
raise ValueError('URL cannot be empty')
if not challenge:
raise ValueError('Challenge cannot be empty')
src_url = parse.urlparse(url)
if src_url.netloc != challenge.source_authority:
raise ValueError('Source URL and Challenge URL do not match')
_lock.acquire()
_cache[src_url.netloc] = challenge
_lock.release()
def clear():
""" Clears the cache. """
_lock.acquire()
_cache.clear() # pylint: disable=redefined-outer-name,unused-variable
_lock.release()
|
