import base64, os, os.path, hmac, json, secrets from datetime import timedelta from expiringdict import ExpiringDict import utils from mailconfig import get_mail_password, get_mail_user_privileges from mfa import get_hash_mfa_state, validate_auth_mfa DEFAULT_KEY_PATH = '/var/lib/mailinabox/api.key' DEFAULT_AUTH_REALM = 'Mail-in-a-Box Management Server' class AuthService: def __init__(self): self.auth_realm = DEFAULT_AUTH_REALM self.key_path = DEFAULT_KEY_PATH self.max_session_duration = timedelta(days=2) self.init_system_api_key() self.sessions = ExpiringDict(max_len=64, max_age_seconds=self.max_session_duration.total_seconds()) def init_system_api_key(self): """Write an API key to a local file so local processes can use the API""" with open(self.key_path, 'r') as file: self.key = file.read() def authenticate(self, request, env, login_only=False, logout=False): """Test if the HTTP Authorization header's username matches the system key, a session key, or if the username/password passed in the header matches a local user. Returns a tuple of the user's email address and list of user privileges (e.g. ('my@email', []) or ('my@email', ['admin']); raises a ValueError on login failure. If the user used the system API key, the user's email is returned as None since this key is not associated with a user.""" def parse_http_authorization_basic(header): def decode(s): return base64.b64decode(s.encode('ascii')).decode('ascii') if " " not in header: return None, None scheme, credentials = header.split(maxsplit=1) if scheme != 'Basic': return None, None credentials = decode(credentials) if ":" not in credentials: return None, None username, password = credentials.split(':', maxsplit=1) return username, password username, password = parse_http_authorization_basic(request.headers.get('Authorization', '')) if username in (None, ""): raise ValueError("Authorization header invalid.") if username.strip() == "" and password.strip() == "": raise ValueError("No email address, password, session key, or API key provided.") # If user passed the system API key, grant administrative privs. This key # is not associated with a user. if username == self.key and not login_only: return (None, ["admin"]) # If the password corresponds with a session token for the user, grant access for that user. if self.get_session(username, password, "login", env) and not login_only: sessionid = password session = self.sessions[sessionid] if logout: # Clear the session. del self.sessions[sessionid] else: # Re-up the session so that it does not expire. self.sessions[sessionid] = session # If no password was given, but a username was given, we're missing some information. elif password.strip() == "": raise ValueError("Enter a password.") else: # The user is trying to log in with a username and a password # (and possibly a MFA token). On failure, an exception is raised. self.check_user_auth(username, password, request, env) # Get privileges for authorization. This call should never fail because by this # point we know the email address is a valid user --- unless the user has been # deleted after the session was granted. On error the call will return a tuple # of an error message and an HTTP status code. privs = get_mail_user_privileges(username, env) if isinstance(privs, tuple): raise ValueError(privs[0]) # Return the authorization information. return (username, privs) def check_user_auth(self, email, pw, request, env): # Validate a user's login email address and password. If MFA is enabled, # check the MFA token in the X-Auth-Token header. # # On login failure, raises a ValueError with a login error message. On # success, nothing is returned. # Authenticate. try: # Get the hashed password of the user. Raise a ValueError if the # email address does not correspond to a user. But wrap it in the # same exception as if a password fails so we don't easily reveal # if an email address is valid. pw_hash = get_mail_password(email, env) # Use 'doveadm pw' to check credentials. doveadm will return # a non-zero exit status if the credentials are no good, # and check_call will raise an exception in that case. utils.shell('check_call', [ "/usr/bin/doveadm", "pw", "-p", pw, "-t", pw_hash, ]) except: # Login failed. raise ValueError("Incorrect email address or password.") # If MFA is enabled, check that MFA passes. status, hints = validate_auth_mfa(email, request, env) if not status: # Login valid. Hints may have more info. raise ValueError(",".join(hints)) def create_user_password_state_token(self, email, env): # Create a token that changes if the user's password or MFA options change # so that sessions become invalid if any of that information changes. msg = get_mail_password(email, env).encode("utf8") # Add to the message the current MFA state, which is a list of MFA information. # Turn it into a string stably. msg += b" " + json.dumps(get_hash_mfa_state(email, env), sort_keys=True).encode("utf8") # Make a HMAC using the system API key as a hash key. hash_key = self.key.encode('ascii') return hmac.new(hash_key, msg, digestmod="sha256").hexdigest() def create_session_key(self, username, env, type=None): # Create a new session. token = secrets.token_hex(32) self.sessions[token] = { "email": username, "password_token": self.create_user_password_state_token(username, env), "type": type, } return token def get_session(self, user_email, session_key, session_type, env): if session_key not in self.sessions: return None session = self.sessions[session_key] if session_type == "login" and session["email"] != user_email: return None if session["type"] != session_type: return None if session["password_token"] != self.create_user_password_state_token(session["email"], env): return None return session