Module connpy.services.user_service

Classes

class UserService (config_dir)

Expand source code

class UserService:
    def __init__(self, config_dir):
        self.config_dir = os.path.abspath(config_dir)
        self.users_dir = os.path.join(self.config_dir, "users")
        self.registry_file = os.path.join(self.users_dir, "registry.yaml")
        
        # Ensure users directory exists
        os.makedirs(self.users_dir, exist_ok=True)
        
        # Reverse index cache: token_hash -> (username, token_id)
        self._token_index: dict[str, tuple[str, str]] = {}

    def _load_registry(self) -> dict:
        """Loads registry from file. If it doesn't exist, initializes it with a new JWT secret."""
        if not os.path.exists(self.registry_file):
            registry = {
                "jwt_secret": secrets.token_hex(32),
                "users": {}
            }
            self._save_registry(registry)
            return registry
        
        try:
            with open(self.registry_file, "r") as f:
                registry = yaml.safe_load(f) or {}
        except Exception:
            registry = {}
            
        if not isinstance(registry, dict):
            registry = {}
            
        if "jwt_secret" not in registry:
            registry["jwt_secret"] = secrets.token_hex(32)
            
        if "users" not in registry or not isinstance(registry["users"], dict):
            registry["users"] = {}
            
        return registry

    def _save_registry(self, data: dict):
        """Safely saves registry structure to registry.yaml."""
        tmp_file = self.registry_file + ".tmp"
        try:
            with open(tmp_file, "w") as f:
                yaml.dump(data, f, default_flow_style=False, sort_keys=False)
            os.replace(tmp_file, self.registry_file)
            os.chmod(self.registry_file, 0o600)
        except Exception as e:
            if os.path.exists(tmp_file):
                try:
                    os.remove(tmp_file)
                except OSError:
                    pass
            raise e

    def _build_token_index(self, registry: dict) -> dict[str, tuple[str, str]]:
        """Builds a reverse index of token_hash -> (username, token_id) for O(1) PAT lookup."""
        index = {}
        for username, user_data in registry.get("users", {}).items():
            for token_id, token_meta in user_data.get("api_tokens", {}).items():
                token_hash = token_meta.get("token_hash")
                if token_hash:
                    index[token_hash] = (username, token_id)
        return index

    def create_user(self, username, password, config_path=None) -> dict:
        """Creates a new user with bcrypt-hashed credentials.
        
        Mode A: config_path=None (fresh user) -> Generates config.yaml and .osk key.
        Mode B: config_path set -> Reuses existing directory after validating its structure.
        """
        if not username or not isinstance(username, str):
            raise ValueError("Username cannot be empty")
            
        if not re.match(r"^[a-zA-Z0-9_-]+$", username):
            raise ValueError("Username must contain only alphanumeric characters, dashes, or underscores")
            
        if not password or not isinstance(password, str):
            raise ValueError("Password cannot be empty")
            
        registry = self._load_registry()
        if username in registry["users"]:
            raise ValueError(f"User '{username}' already exists")
            
        # Resolve path and initialize configuration
        if config_path is None:
            user_dir = os.path.join(self.users_dir, username)
            os.makedirs(user_dir, exist_ok=True)
            
            # Create subdirs for plugins and sessions
            os.makedirs(os.path.join(user_dir, "plugins"), exist_ok=True)
            os.makedirs(os.path.join(user_dir, "ai_sessions"), exist_ok=True)
            
            # Create default config.yaml & .osk key via configfile
            conf_file = os.path.join(user_dir, "config.yaml")
            configfile(conf=conf_file)
            
            stored_config_path = None
        else:
            abs_config_path = os.path.abspath(config_path)
            os.makedirs(abs_config_path, exist_ok=True)
            
            # Create subdirs for plugins and sessions in the custom path
            os.makedirs(os.path.join(abs_config_path, "plugins"), exist_ok=True)
            os.makedirs(os.path.join(abs_config_path, "ai_sessions"), exist_ok=True)
            
            # Create default config.yaml & .osk key via configfile if config.yaml is not present
            conf_file = os.path.join(abs_config_path, "config.yaml")
            if not os.path.exists(conf_file):
                configfile(conf=conf_file)
                
            stored_config_path = abs_config_path

        # Hash password securely
        password_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
        
        user_entry = {
            "password_hash": password_hash,
            "config_path": stored_config_path,
            "created": datetime.datetime.now(datetime.timezone.utc).isoformat()
        }
        
        registry["users"][username] = user_entry
        self._save_registry(registry)
        
        return {
            "username": username,
            "config_path": stored_config_path,
            "created": user_entry["created"]
        }

    def delete_user(self, username):
        """Removes user from the registry and cleans up config directory if server-managed."""
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
            
        user_data = registry["users"][username]
        config_path = user_data.get("config_path")
        
        if config_path is None:
            user_dir = os.path.join(self.users_dir, username)
            if os.path.exists(user_dir):
                shutil.rmtree(user_dir, ignore_errors=True)
                
        del registry["users"][username]
        self._save_registry(registry)

    def list_users(self) -> list[dict]:
        """Lists all registered users with metadata."""
        registry = self._load_registry()
        return [
            {
                "username": name,
                "config_path": data.get("config_path"),
                "created": data.get("created")
            }
            for name, data in registry.get("users", {}).items()
        ]

    def get_user(self, username) -> dict:
        """Retrieves raw metadata for a specific user."""
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
            
        data = registry["users"][username]
        return {
            "username": username,
            "config_path": data.get("config_path"),
            "created": data.get("created"),
            "password_hash": data.get("password_hash")
        }

    def change_password(self, username, old_password, new_password):
        """Verifies old password and updates registry with new hashed password."""
        if not new_password or not isinstance(new_password, str):
            raise ValueError("New password cannot be empty")
            
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
            
        user_data = registry["users"][username]
        if not bcrypt.checkpw(old_password.encode("utf-8"), user_data["password_hash"].encode("utf-8")):
            raise ValueError("Invalid credentials")
            
        # Update hash
        user_data["password_hash"] = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
        self._save_registry(registry)

    def admin_change_password(self, username, new_password):
        """Administrative password override (does not require old password)."""
        if not new_password or not isinstance(new_password, str):
            raise ValueError("New password cannot be empty")
            
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
            
        user_data = registry["users"][username]
        user_data["password_hash"] = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
        self._save_registry(registry)

    def authenticate(self, username, password) -> bool:
        """Verifies if the credentials are valid using bcrypt."""
        registry = self._load_registry()
        if username not in registry["users"]:
            return False
            
        user_data = registry["users"][username]
        return bcrypt.checkpw(password.encode("utf-8"), user_data["password_hash"].encode("utf-8"))

    def generate_jwt(self, username) -> str:
        """Generates a secure JSON Web Token for the user expiring in 12 hours."""
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
            
        expiration = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=12)
        payload = {
            "sub": username,
            "exp": expiration
        }
        
        secret = os.environ.get("CONNPY_JWT_SECRET") or registry["jwt_secret"]
        token = jwt.encode(payload, secret, algorithm="HS256")
        if isinstance(token, bytes):
            token = token.decode("utf-8")
            
        return token

    def verify_jwt(self, token) -> str | None:
        """Decodes JWT and returns username if token is valid and unexpired."""
        registry = self._load_registry()
        try:
            secret = os.environ.get("CONNPY_JWT_SECRET") or registry["jwt_secret"]
            payload = jwt.decode(token, secret, algorithms=["HS256"])
            return payload.get("sub")
        except (jwt.ExpiredSignatureError, jwt.InvalidTokenError, KeyError):
            return None

    # --- Personal Access Token (PAT) Management ---

    def create_api_token(self, username: str, name: str, expires_in_days: int | None = None) -> dict:
        """Creates a Personal Access Token for the user.
        
        Returns the raw token ONCE. Only the SHA-256 hash is persisted.
        """
        if not name or not isinstance(name, str):
            raise ValueError("Token name cannot be empty")
        
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
        
        user_data = registry["users"][username]
        if "api_tokens" not in user_data:
            user_data["api_tokens"] = {}
        
        # Generate cryptographically secure token with recognizable prefix
        raw_secret = secrets.token_hex(32)
        raw_token = f"cnp_pat_{raw_secret}"
        token_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
        token_id = f"tok_{secrets.token_hex(4)}"
        
        now = datetime.datetime.now(datetime.timezone.utc)
        expires_at = None
        if expires_in_days and expires_in_days > 0:
            expires_at = (now + datetime.timedelta(days=expires_in_days)).isoformat()
        
        user_data["api_tokens"][token_id] = {
            "name": name,
            "token_hash": token_hash,
            "token_prefix": raw_token[:16],
            "created_at": now.isoformat(),
            "last_used_at": None,
            "expires_at": expires_at,
        }
        
        self._save_registry(registry)
        self._token_index = self._build_token_index(registry)
        
        return {
            "token_id": token_id,
            "raw_token": raw_token,
            "name": name,
        }

    def list_api_tokens(self, username: str) -> list[dict]:
        """Lists all active API tokens for a user (without sensitive data)."""
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
        
        tokens = registry["users"][username].get("api_tokens", {})
        return [
            {
                "token_id": tid,
                "name": meta.get("name"),
                "token_prefix": meta.get("token_prefix"),
                "created_at": meta.get("created_at"),
                "last_used_at": meta.get("last_used_at"),
                "expires_at": meta.get("expires_at"),
            }
            for tid, meta in tokens.items()
        ]

    def revoke_api_token(self, username: str, token_id: str) -> bool:
        """Revokes (deletes) a specific API token. Returns True if found and removed."""
        registry = self._load_registry()
        if username not in registry["users"]:
            raise ValueError(f"User '{username}' not found")
        
        tokens = registry["users"][username].get("api_tokens", {})
        if token_id not in tokens:
            return False
        
        del tokens[token_id]
        self._save_registry(registry)
        self._token_index = self._build_token_index(registry)
        return True

    def verify_api_token(self, raw_token: str) -> str | None:
        """Validates a PAT by hashing it and looking up the reverse index.
        
        Returns username if valid and not expired, None otherwise.
        """
        token_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
        
        # Rebuild index if empty (cold start or after process restart)
        if not self._token_index:
            registry = self._load_registry()
            self._token_index = self._build_token_index(registry)
        
        match = self._token_index.get(token_hash)
        if not match:
            return None
        
        username, token_id = match
        
        # Validate token still exists and check expiration
        registry = self._load_registry()
        user_data = registry.get("users", {}).get(username, {})
        token_meta = user_data.get("api_tokens", {}).get(token_id)
        
        if not token_meta:
            # Token was revoked between index build and now
            self._token_index = self._build_token_index(registry)
            return None
        
        # Check expiration
        expires_at = token_meta.get("expires_at")
        if expires_at:
            exp_dt = datetime.datetime.fromisoformat(expires_at)
            if datetime.datetime.now(datetime.timezone.utc) > exp_dt:
                return None
        
        # Update last_used_at
        token_meta["last_used_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
        self._save_registry(registry)
        
        return username

Methods

def admin_change_password(self, username, new_password)

Expand source code

def admin_change_password(self, username, new_password):
    """Administrative password override (does not require old password)."""
    if not new_password or not isinstance(new_password, str):
        raise ValueError("New password cannot be empty")
        
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
        
    user_data = registry["users"][username]
    user_data["password_hash"] = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
    self._save_registry(registry)

Administrative password override (does not require old password).

def authenticate(self, username, password) ‑> bool

Expand source code

def authenticate(self, username, password) -> bool:
    """Verifies if the credentials are valid using bcrypt."""
    registry = self._load_registry()
    if username not in registry["users"]:
        return False
        
    user_data = registry["users"][username]
    return bcrypt.checkpw(password.encode("utf-8"), user_data["password_hash"].encode("utf-8"))

Verifies if the credentials are valid using bcrypt.

def change_password(self, username, old_password, new_password)

Expand source code

def change_password(self, username, old_password, new_password):
    """Verifies old password and updates registry with new hashed password."""
    if not new_password or not isinstance(new_password, str):
        raise ValueError("New password cannot be empty")
        
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
        
    user_data = registry["users"][username]
    if not bcrypt.checkpw(old_password.encode("utf-8"), user_data["password_hash"].encode("utf-8")):
        raise ValueError("Invalid credentials")
        
    # Update hash
    user_data["password_hash"] = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
    self._save_registry(registry)

Verifies old password and updates registry with new hashed password.

def create_api_token(self, username: str, name: str, expires_in_days: int | None = None) ‑> dict

Expand source code

def create_api_token(self, username: str, name: str, expires_in_days: int | None = None) -> dict:
    """Creates a Personal Access Token for the user.
    
    Returns the raw token ONCE. Only the SHA-256 hash is persisted.
    """
    if not name or not isinstance(name, str):
        raise ValueError("Token name cannot be empty")
    
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
    
    user_data = registry["users"][username]
    if "api_tokens" not in user_data:
        user_data["api_tokens"] = {}
    
    # Generate cryptographically secure token with recognizable prefix
    raw_secret = secrets.token_hex(32)
    raw_token = f"cnp_pat_{raw_secret}"
    token_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
    token_id = f"tok_{secrets.token_hex(4)}"
    
    now = datetime.datetime.now(datetime.timezone.utc)
    expires_at = None
    if expires_in_days and expires_in_days > 0:
        expires_at = (now + datetime.timedelta(days=expires_in_days)).isoformat()
    
    user_data["api_tokens"][token_id] = {
        "name": name,
        "token_hash": token_hash,
        "token_prefix": raw_token[:16],
        "created_at": now.isoformat(),
        "last_used_at": None,
        "expires_at": expires_at,
    }
    
    self._save_registry(registry)
    self._token_index = self._build_token_index(registry)
    
    return {
        "token_id": token_id,
        "raw_token": raw_token,
        "name": name,
    }

Creates a Personal Access Token for the user.

Returns the raw token ONCE. Only the SHA-256 hash is persisted.

def create_user(self, username, password, config_path=None) ‑> dict

Expand source code

def create_user(self, username, password, config_path=None) -> dict:
    """Creates a new user with bcrypt-hashed credentials.
    
    Mode A: config_path=None (fresh user) -> Generates config.yaml and .osk key.
    Mode B: config_path set -> Reuses existing directory after validating its structure.
    """
    if not username or not isinstance(username, str):
        raise ValueError("Username cannot be empty")
        
    if not re.match(r"^[a-zA-Z0-9_-]+$", username):
        raise ValueError("Username must contain only alphanumeric characters, dashes, or underscores")
        
    if not password or not isinstance(password, str):
        raise ValueError("Password cannot be empty")
        
    registry = self._load_registry()
    if username in registry["users"]:
        raise ValueError(f"User '{username}' already exists")
        
    # Resolve path and initialize configuration
    if config_path is None:
        user_dir = os.path.join(self.users_dir, username)
        os.makedirs(user_dir, exist_ok=True)
        
        # Create subdirs for plugins and sessions
        os.makedirs(os.path.join(user_dir, "plugins"), exist_ok=True)
        os.makedirs(os.path.join(user_dir, "ai_sessions"), exist_ok=True)
        
        # Create default config.yaml & .osk key via configfile
        conf_file = os.path.join(user_dir, "config.yaml")
        configfile(conf=conf_file)
        
        stored_config_path = None
    else:
        abs_config_path = os.path.abspath(config_path)
        os.makedirs(abs_config_path, exist_ok=True)
        
        # Create subdirs for plugins and sessions in the custom path
        os.makedirs(os.path.join(abs_config_path, "plugins"), exist_ok=True)
        os.makedirs(os.path.join(abs_config_path, "ai_sessions"), exist_ok=True)
        
        # Create default config.yaml & .osk key via configfile if config.yaml is not present
        conf_file = os.path.join(abs_config_path, "config.yaml")
        if not os.path.exists(conf_file):
            configfile(conf=conf_file)
            
        stored_config_path = abs_config_path

    # Hash password securely
    password_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
    
    user_entry = {
        "password_hash": password_hash,
        "config_path": stored_config_path,
        "created": datetime.datetime.now(datetime.timezone.utc).isoformat()
    }
    
    registry["users"][username] = user_entry
    self._save_registry(registry)
    
    return {
        "username": username,
        "config_path": stored_config_path,
        "created": user_entry["created"]
    }

Creates a new user with bcrypt-hashed credentials.

Mode A: config_path=None (fresh user) -> Generates config.yaml and .osk key. Mode B: config_path set -> Reuses existing directory after validating its structure.

def delete_user(self, username)

Expand source code

def delete_user(self, username):
    """Removes user from the registry and cleans up config directory if server-managed."""
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
        
    user_data = registry["users"][username]
    config_path = user_data.get("config_path")
    
    if config_path is None:
        user_dir = os.path.join(self.users_dir, username)
        if os.path.exists(user_dir):
            shutil.rmtree(user_dir, ignore_errors=True)
            
    del registry["users"][username]
    self._save_registry(registry)

Removes user from the registry and cleans up config directory if server-managed.

def generate_jwt(self, username) ‑> str

Expand source code

def generate_jwt(self, username) -> str:
    """Generates a secure JSON Web Token for the user expiring in 12 hours."""
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
        
    expiration = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=12)
    payload = {
        "sub": username,
        "exp": expiration
    }
    
    secret = os.environ.get("CONNPY_JWT_SECRET") or registry["jwt_secret"]
    token = jwt.encode(payload, secret, algorithm="HS256")
    if isinstance(token, bytes):
        token = token.decode("utf-8")
        
    return token

Generates a secure JSON Web Token for the user expiring in 12 hours.

def get_user(self, username) ‑> dict

Expand source code

def get_user(self, username) -> dict:
    """Retrieves raw metadata for a specific user."""
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
        
    data = registry["users"][username]
    return {
        "username": username,
        "config_path": data.get("config_path"),
        "created": data.get("created"),
        "password_hash": data.get("password_hash")
    }

Retrieves raw metadata for a specific user.

def list_api_tokens(self, username: str) ‑> list[dict]

Expand source code

def list_api_tokens(self, username: str) -> list[dict]:
    """Lists all active API tokens for a user (without sensitive data)."""
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
    
    tokens = registry["users"][username].get("api_tokens", {})
    return [
        {
            "token_id": tid,
            "name": meta.get("name"),
            "token_prefix": meta.get("token_prefix"),
            "created_at": meta.get("created_at"),
            "last_used_at": meta.get("last_used_at"),
            "expires_at": meta.get("expires_at"),
        }
        for tid, meta in tokens.items()
    ]

Lists all active API tokens for a user (without sensitive data).

def list_users(self) ‑> list[dict]

Expand source code

def list_users(self) -> list[dict]:
    """Lists all registered users with metadata."""
    registry = self._load_registry()
    return [
        {
            "username": name,
            "config_path": data.get("config_path"),
            "created": data.get("created")
        }
        for name, data in registry.get("users", {}).items()
    ]

Lists all registered users with metadata.

def revoke_api_token(self, username: str, token_id: str) ‑> bool

Expand source code

def revoke_api_token(self, username: str, token_id: str) -> bool:
    """Revokes (deletes) a specific API token. Returns True if found and removed."""
    registry = self._load_registry()
    if username not in registry["users"]:
        raise ValueError(f"User '{username}' not found")
    
    tokens = registry["users"][username].get("api_tokens", {})
    if token_id not in tokens:
        return False
    
    del tokens[token_id]
    self._save_registry(registry)
    self._token_index = self._build_token_index(registry)
    return True

Revokes (deletes) a specific API token. Returns True if found and removed.

def verify_api_token(self, raw_token: str) ‑> str | None

Expand source code

def verify_api_token(self, raw_token: str) -> str | None:
    """Validates a PAT by hashing it and looking up the reverse index.
    
    Returns username if valid and not expired, None otherwise.
    """
    token_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
    
    # Rebuild index if empty (cold start or after process restart)
    if not self._token_index:
        registry = self._load_registry()
        self._token_index = self._build_token_index(registry)
    
    match = self._token_index.get(token_hash)
    if not match:
        return None
    
    username, token_id = match
    
    # Validate token still exists and check expiration
    registry = self._load_registry()
    user_data = registry.get("users", {}).get(username, {})
    token_meta = user_data.get("api_tokens", {}).get(token_id)
    
    if not token_meta:
        # Token was revoked between index build and now
        self._token_index = self._build_token_index(registry)
        return None
    
    # Check expiration
    expires_at = token_meta.get("expires_at")
    if expires_at:
        exp_dt = datetime.datetime.fromisoformat(expires_at)
        if datetime.datetime.now(datetime.timezone.utc) > exp_dt:
            return None
    
    # Update last_used_at
    token_meta["last_used_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
    self._save_registry(registry)
    
    return username

Validates a PAT by hashing it and looking up the reverse index.

Returns username if valid and not expired, None otherwise.

def verify_jwt(self, token) ‑> str | None

Expand source code

def verify_jwt(self, token) -> str | None:
    """Decodes JWT and returns username if token is valid and unexpired."""
    registry = self._load_registry()
    try:
        secret = os.environ.get("CONNPY_JWT_SECRET") or registry["jwt_secret"]
        payload = jwt.decode(token, secret, algorithms=["HS256"])
        return payload.get("sub")
    except (jwt.ExpiredSignatureError, jwt.InvalidTokenError, KeyError):
        return None

Decodes JWT and returns username if token is valid and unexpired.