Compare commits
No commits in common. "9d0d0b2686a78a7d3f13aefc7dd00e5f97c287bb" and "f3b525d715597f81d455f440691095ff73ba3f79" have entirely different histories.
9d0d0b2686
...
f3b525d715
8
.gitignore
vendored
8
.gitignore
vendored
@ -23,13 +23,11 @@ dist
|
||||
.pytest_cache
|
||||
venv
|
||||
|
||||
flask_session
|
||||
instance
|
||||
|
||||
.DS_Store
|
||||
.idea
|
||||
|
||||
instance
|
||||
flask_session
|
||||
|
||||
.scannerwork
|
||||
sonar-project.properties
|
||||
|
||||
excludes
|
||||
|
@ -61,16 +61,6 @@ def calc_response(
|
||||
cnonce or nc is missing with the auth or auth-int qop.
|
||||
"""
|
||||
|
||||
def validate_required(field: t.Optional[str], error: str) -> None:
|
||||
"""Validates a required field.
|
||||
|
||||
:param field: The field that is required.
|
||||
:param error: The error message.
|
||||
:return: None.
|
||||
"""
|
||||
if field is None:
|
||||
raise UnauthorizedException(error)
|
||||
|
||||
def calc_ha1() -> str:
|
||||
"""Calculates and returns the first hash.
|
||||
|
||||
@ -78,13 +68,16 @@ def calc_response(
|
||||
:raise UnauthorizedException: When the cnonce is missing with the MD5-sess
|
||||
algorithm.
|
||||
"""
|
||||
if algorithm is None or algorithm == "MD5":
|
||||
return password_hash
|
||||
if algorithm == "MD5-sess":
|
||||
validate_required(
|
||||
cnonce, f"Missing \"cnonce\" with algorithm=\"{algorithm}\"")
|
||||
if cnonce is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"cnonce\" with algorithm=\"{algorithm}\"")
|
||||
return md5(f"{password_hash}:{nonce}:{cnonce}".encode("utf8")) \
|
||||
.hexdigest()
|
||||
# algorithm is None or algorithm == "MD5"
|
||||
return password_hash
|
||||
raise UnauthorizedException(
|
||||
f"Unsupported algorithm=\"{algorithm}\"")
|
||||
|
||||
def calc_ha2() -> str:
|
||||
"""Calculates the second hash.
|
||||
@ -93,20 +86,30 @@ def calc_response(
|
||||
:raise UnauthorizedException: When the body is missing with
|
||||
qop="auth-int".
|
||||
"""
|
||||
if qop is None or qop == "auth":
|
||||
return md5(f"{method}:{uri}".encode("utf8")).hexdigest()
|
||||
if qop == "auth-int":
|
||||
validate_required(body, f"Missing \"body\" with qop=\"{qop}\"")
|
||||
if body is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"body\" with qop=\"{qop}\"")
|
||||
return md5(
|
||||
f"{method}:{uri}:{md5(body).hexdigest()}".encode("utf8")) \
|
||||
.hexdigest()
|
||||
# qop is None or qop == "auth"
|
||||
return md5(f"{method}:{uri}".encode("utf8")).hexdigest()
|
||||
raise UnauthorizedException(f"Unsupported qop=\"{qop}\"")
|
||||
|
||||
ha1: str = calc_ha1()
|
||||
ha2: str = calc_ha2()
|
||||
if qop is None:
|
||||
return md5(f"{ha1}:{nonce}:{ha2}".encode("utf8")).hexdigest()
|
||||
if qop == "auth" or qop == "auth-int":
|
||||
validate_required(cnonce, f"Missing \"cnonce\" with the qop=\"{qop}\"")
|
||||
validate_required(nc, f"Missing \"nc\" with the qop=\"{qop}\"")
|
||||
if cnonce is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"cnonce\" with the qop=\"{qop}\"")
|
||||
if nc is None:
|
||||
raise UnauthorizedException(
|
||||
f"Missing \"nc\" with the qop=\"{qop}\"")
|
||||
return md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode("utf8"))\
|
||||
.hexdigest()
|
||||
# qop is None
|
||||
return md5(f"{ha1}:{nonce}:{ha2}".encode("utf8")).hexdigest()
|
||||
if cnonce is None:
|
||||
raise UnauthorizedException(
|
||||
f"Unsupported qop=\"{qop}\"")
|
||||
|
@ -110,36 +110,6 @@ class DigestAuth:
|
||||
class NoLogInException(Exception):
|
||||
"""The exception thrown when the user is not authorized."""
|
||||
|
||||
def get_logged_in_user() -> t.Optional[t.Any]:
|
||||
"""Returns the currently logged-in user.
|
||||
|
||||
:return: The currently logged-in user.
|
||||
:raise NoLogInException: When the user is not logged in.
|
||||
"""
|
||||
if "user" not in session:
|
||||
raise NoLogInException
|
||||
user: t.Optional[t.Any] = self.__get_user(session["user"])
|
||||
if user is None:
|
||||
raise NoLogInException
|
||||
return user
|
||||
|
||||
def auth_user(state: AuthState) -> t.Any:
|
||||
"""Authenticates a user.
|
||||
|
||||
:param state: The authentication state.
|
||||
:return: The user.
|
||||
:raise UnauthorizedException: When the authentication fails.
|
||||
"""
|
||||
authorization: Authorization = request.authorization
|
||||
if authorization is None:
|
||||
raise UnauthorizedException
|
||||
if authorization.type != "digest":
|
||||
raise UnauthorizedException(
|
||||
"Not an HTTP digest authorization")
|
||||
self.authenticate(state)
|
||||
session["user"] = authorization.username
|
||||
return self.__get_user(authorization.username)
|
||||
|
||||
@wraps(view)
|
||||
def login_required_view(*args, **kwargs) -> t.Any:
|
||||
"""The login-protected view.
|
||||
@ -149,24 +119,36 @@ class DigestAuth:
|
||||
:return: The response.
|
||||
"""
|
||||
try:
|
||||
g.user = get_logged_in_user()
|
||||
if "user" not in session:
|
||||
raise NoLogInException
|
||||
user: t.Optional[t.Any] = self.__get_user(session["user"])
|
||||
if user is None:
|
||||
raise NoLogInException
|
||||
g.user = user
|
||||
return view(*args, **kwargs)
|
||||
except NoLogInException:
|
||||
pass
|
||||
|
||||
state: AuthState = AuthState()
|
||||
try:
|
||||
g.user = auth_user(state)
|
||||
self.__on_login(g.user)
|
||||
return view(*args, **kwargs)
|
||||
except UnauthorizedException as e:
|
||||
if len(e.args) > 0:
|
||||
sys.stderr.write(e.args[0] + "\n")
|
||||
response: Response = Response()
|
||||
response.status = 401
|
||||
response.headers["WWW-Authenticate"] \
|
||||
= self.make_response_header(state)
|
||||
abort(response)
|
||||
state: AuthState = AuthState()
|
||||
authorization: Authorization = request.authorization
|
||||
try:
|
||||
if authorization is None:
|
||||
raise UnauthorizedException
|
||||
if authorization.type != "digest":
|
||||
raise UnauthorizedException(
|
||||
"Not an HTTP digest authorization")
|
||||
self.authenticate(state)
|
||||
session["user"] = authorization.username
|
||||
user = self.__get_user(authorization.username)
|
||||
g.user = user
|
||||
self.__on_login(user)
|
||||
return view(*args, **kwargs)
|
||||
except UnauthorizedException as e:
|
||||
if len(e.args) > 0:
|
||||
sys.stderr.write(e.args[0] + "\n")
|
||||
response: Response = Response()
|
||||
response.status = 401
|
||||
response.headers["WWW-Authenticate"] \
|
||||
= self.make_response_header(state)
|
||||
abort(response)
|
||||
|
||||
return login_required_view
|
||||
|
||||
@ -221,19 +203,9 @@ class DigestAuth:
|
||||
:param state: The authorization state.
|
||||
:return: The WWW-Authenticate response header.
|
||||
"""
|
||||
|
||||
def get_opaque() -> t.Optional[str]:
|
||||
"""Returns the opaque value.
|
||||
|
||||
:return: The opaque value.
|
||||
"""
|
||||
if not self.use_opaque:
|
||||
return None
|
||||
if state.opaque is not None:
|
||||
return state.opaque
|
||||
return self.serializer.dumps(randbits(32), salt="opaque")
|
||||
|
||||
opaque: t.Optional[str] = get_opaque()
|
||||
opaque: t.Optional[str] = None if not self.use_opaque else \
|
||||
(state.opaque if state.opaque is not None
|
||||
else self.serializer.dumps(randbits(32), salt="opaque"))
|
||||
nonce: str = self.serializer.dumps(
|
||||
randbits(32),
|
||||
salt="nonce" if opaque is None else f"nonce-{opaque}")
|
||||
|
@ -66,8 +66,9 @@ class Client(WerkzeugClient):
|
||||
:return: The request authorization.
|
||||
"""
|
||||
qop: t.Optional[t.Literal["auth", "auth-int"]] = None
|
||||
if www_authenticate.qop is not None and "auth" in www_authenticate.qop:
|
||||
qop = "auth"
|
||||
if www_authenticate.qop is not None:
|
||||
if "auth" in www_authenticate.qop:
|
||||
qop = "auth"
|
||||
|
||||
cnonce: t.Optional[str] = None
|
||||
if qop is not None or www_authenticate.algorithm == "MD5-sess":
|
||||
|
Loading…
Reference in New Issue
Block a user