Skip to content

ccproxy.auth.storage

ccproxy.auth.storage

Token storage implementations for authentication.

TokenStorage

Bases: ABC

Abstract interface for token storage operations.

load abstractmethod async 

load()

Load credentials from storage.

Returns:

Type Description
ClaudeCredentials | None

Parsed credentials if found and valid, None otherwise
Source code in ccproxy/auth/storage/base.py 
11
12
13
14
15
16
17
18
@abstractmethod
async def load(self) -> ClaudeCredentials | None:
    """Load credentials from storage.

    Returns:
        Parsed credentials if found and valid, None otherwise
    """
    pass

save abstractmethod async 

save(credentials)

Save credentials to storage.

Parameters:

Name Type Description Default
credentials ClaudeCredentials

Credentials to save

 required

Returns:

Type Description
bool

True if saved successfully, False otherwise
Source code in ccproxy/auth/storage/base.py 
20
21
22
23
24
25
26
27
28
29
30
@abstractmethod
async def save(self, credentials: ClaudeCredentials) -> bool:
    """Save credentials to storage.

    Args:
        credentials: Credentials to save

    Returns:
        True if saved successfully, False otherwise
    """
    pass

exists abstractmethod async 

exists()

Check if credentials exist in storage.

Returns:

Type Description
bool

True if credentials exist, False otherwise
Source code in ccproxy/auth/storage/base.py 
32
33
34
35
36
37
38
39
@abstractmethod
async def exists(self) -> bool:
    """Check if credentials exist in storage.

    Returns:
        True if credentials exist, False otherwise
    """
    pass

delete abstractmethod async 

delete()

Delete credentials from storage.

Returns:

Type Description
bool

True if deleted successfully, False otherwise
Source code in ccproxy/auth/storage/base.py 
41
42
43
44
45
46
47
48
@abstractmethod
async def delete(self) -> bool:
    """Delete credentials from storage.

    Returns:
        True if deleted successfully, False otherwise
    """
    pass

get_location abstractmethod 

get_location()

Get the storage location description.

Returns:

Type Description
str

Human-readable description of where credentials are stored
Source code in ccproxy/auth/storage/base.py 
50
51
52
53
54
55
56
57
@abstractmethod
def get_location(self) -> str:
    """Get the storage location description.

    Returns:
        Human-readable description of where credentials are stored
    """
    pass

JsonFileTokenStorage 

JsonFileTokenStorage(file_path)

Bases: TokenStorage

JSON file storage implementation for Claude credentials with keyring fallback.

Parameters:

Name Type Description Default
file_path Path

Path to the JSON credentials file

 required
Source code in ccproxy/auth/storage/json_file.py 
24
25
26
27
28
29
30
def __init__(self, file_path: Path):
    """Initialize JSON file storage.

    Args:
        file_path: Path to the JSON credentials file
    """
    self.file_path = file_path

load async 

load()

Load credentials from JSON file .

Returns:

Type Description
ClaudeCredentials | None

Parsed credentials if found and valid, None otherwise

Raises:

Type Description
CredentialsInvalidError

If the JSON file is invalid
CredentialsStorageError

If there's an error reading the file
Source code in ccproxy/auth/storage/json_file.py 
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
async def load(self) -> ClaudeCredentials | None:
    """Load credentials from JSON file .

    Returns:
        Parsed credentials if found and valid, None otherwise

    Raises:
        CredentialsInvalidError: If the JSON file is invalid
        CredentialsStorageError: If there's an error reading the file
    """
    if not await self.exists():
        logger.debug("credentials_file_not_found", path=str(self.file_path))
        return None

    try:
        logger.debug(
            "credentials_load_start", source="file", path=str(self.file_path)
        )
        with self.file_path.open() as f:
            data = json.load(f)

        credentials = ClaudeCredentials.model_validate(data)
        logger.debug("credentials_load_completed", source="file")

        return credentials

    except json.JSONDecodeError as e:
        raise CredentialsInvalidError(
            f"Failed to parse credentials file {self.file_path}: {e}"
        ) from e
    except Exception as e:
        raise CredentialsStorageError(
            f"Error loading credentials from {self.file_path}: {e}"
        ) from e

save async 

save(credentials)

Save credentials to both keyring and JSON file.

Parameters:

Name Type Description Default
credentials ClaudeCredentials

Credentials to save

 required

Returns:

Type Description
bool

True if saved successfully, False otherwise

Raises:

Type Description
CredentialsStorageError

If there's an error writing the file
Source code in ccproxy/auth/storage/json_file.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def save(self, credentials: ClaudeCredentials) -> bool:
    """Save credentials to both keyring and JSON file.

    Args:
        credentials: Credentials to save

    Returns:
        True if saved successfully, False otherwise

    Raises:
        CredentialsStorageError: If there's an error writing the file
    """
    try:
        # Convert to dict with proper aliases
        data = credentials.model_dump(by_alias=True, mode="json")

        # Always save to file as well
        # Ensure parent directory exists
        self.file_path.parent.mkdir(parents=True, exist_ok=True)

        # Use atomic write: write to temp file then rename
        temp_path = self.file_path.with_suffix(".tmp")

        try:
            with temp_path.open("w") as f:
                json.dump(data, f, indent=2)

            # Set appropriate file permissions (read/write for owner only)
            temp_path.chmod(0o600)

            # Atomically replace the original file
            Path.replace(temp_path, self.file_path)

            logger.debug(
                "credentials_save_completed",
                source="file",
                path=str(self.file_path),
            )
            return True
        except Exception as e:
            raise
        finally:
            # Clean up temp file if it exists
            if temp_path.exists():
                with contextlib.suppress(Exception):
                    temp_path.unlink()

    except Exception as e:
        raise CredentialsStorageError(f"Error saving credentials: {e}") from e

exists async 

exists()

Check if credentials file exists.

Returns:

Type Description
bool

True if file exists, False otherwise
Source code in ccproxy/auth/storage/json_file.py 
117
118
119
120
121
122
123
async def exists(self) -> bool:
    """Check if credentials file exists.

    Returns:
        True if file exists, False otherwise
    """
    return self.file_path.exists() and self.file_path.is_file()

delete async 

delete()

Delete credentials from both keyring and file.

Returns:

Type Description
bool

True if deleted successfully, False otherwise

Raises:

Type Description
CredentialsStorageError

If there's an error deleting the file
Source code in ccproxy/auth/storage/json_file.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
async def delete(self) -> bool:
    """Delete credentials from both keyring and file.

    Returns:
        True if deleted successfully, False otherwise

    Raises:
        CredentialsStorageError: If there's an error deleting the file
    """
    deleted = False

    # Delete from file
    try:
        if await self.exists():
            self.file_path.unlink()
            logger.debug(
                "credentials_delete_completed",
                source="file",
                path=str(self.file_path),
            )
            deleted = True
    except Exception as e:
        if not deleted:  # Only raise if we failed to delete from both
            raise CredentialsStorageError(f"Error deleting credentials: {e}") from e
        logger.debug("credentials_delete_partial", source="file", error=str(e))

    return deleted

get_location 

get_location()

Get the storage location description.

Returns:

Type Description
str

Path to the JSON file with keyring info if available
Source code in ccproxy/auth/storage/json_file.py 
153
154
155
156
157
158
159
def get_location(self) -> str:
    """Get the storage location description.

    Returns:
        Path to the JSON file with keyring info if available
    """
    return str(self.file_path)

KeyringTokenStorage 

KeyringTokenStorage(
    service_name="claude-code-proxy", username="default"
)

Bases: TokenStorage

OS keyring storage implementation for Claude credentials.

Parameters:

Name Type Description Default
service_name str

Name of the service in the keyring

 'claude-code-proxy'
username str

Username to associate with the stored credentials

 'default'
Source code in ccproxy/auth/storage/keyring.py 
23
24
25
26
27
28
29
30
31
32
33
def __init__(
    self, service_name: str = "claude-code-proxy", username: str = "default"
):
    """Initialize keyring storage.

    Args:
        service_name: Name of the service in the keyring
        username: Username to associate with the stored credentials
    """
    self.service_name = service_name
    self.username = username

load async 

load()

Load credentials from the OS keyring.

Returns:

Type Description
ClaudeCredentials | None

Parsed credentials if found and valid, None otherwise

Raises:

Type Description
CredentialsStorageError

If the stored data is invalid
CredentialsStorageError

If there's an error reading from keyring
Source code in ccproxy/auth/storage/keyring.py 
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
78
79
80
81
82
83
async def load(self) -> ClaudeCredentials | None:
    """Load credentials from the OS keyring.

    Returns:
        Parsed credentials if found and valid, None otherwise

    Raises:
        CredentialsStorageError: If the stored data is invalid
        CredentialsStorageError: If there's an error reading from keyring
    """
    try:
        import keyring
    except ImportError as e:
        raise CredentialsStorageError(
            "keyring package is required for keyring storage. "
            "Install it with: pip install keyring"
        ) from e

    try:
        logger.debug(
            "credentials_load_start",
            source="keyring",
            service_name=self.service_name,
        )
        password = keyring.get_password(self.service_name, self.username)

        if password is None:
            logger.debug(
                "credentials_not_found",
                source="keyring",
                service_name=self.service_name,
            )
            return None

        # Parse the stored JSON
        data = json.loads(password)
        credentials = ClaudeCredentials.model_validate(data)

        self._log_credential_details(credentials)
        return credentials

    except json.JSONDecodeError as e:
        raise CredentialsStorageError(
            f"Failed to parse credentials from keyring: {e}"
        ) from e
    except Exception as e:
        raise CredentialsStorageError(
            f"Error loading credentials from keyring: {e}"
        ) from e

save async 

save(credentials)

Save credentials to the OS keyring.

Parameters:

Name Type Description Default
credentials ClaudeCredentials

Credentials to save

 required

Returns:

Type Description
bool

True if saved successfully, False otherwise

Raises:

Type Description
CredentialsStorageError

If there's an error writing to keyring
Source code in ccproxy/auth/storage/keyring.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def save(self, credentials: ClaudeCredentials) -> bool:
    """Save credentials to the OS keyring.

    Args:
        credentials: Credentials to save

    Returns:
        True if saved successfully, False otherwise

    Raises:
        CredentialsStorageError: If there's an error writing to keyring
    """
    try:
        import keyring
    except ImportError as e:
        raise CredentialsStorageError(
            "keyring package is required for keyring storage. "
            "Install it with: pip install keyring"
        ) from e

    try:
        # Convert to JSON string
        data = credentials.model_dump(by_alias=True)
        json_data = json.dumps(data)

        # Store in keyring
        keyring.set_password(self.service_name, self.username, json_data)

        logger.debug(
            "credentials_save_completed",
            source="keyring",
            service_name=self.service_name,
        )
        return True

    except Exception as e:
        raise CredentialsStorageError(
            f"Error saving credentials to keyring: {e}"
        ) from e

exists async 

exists()

Check if credentials exist in the keyring.

Returns:

Type Description
bool

True if credentials exist, False otherwise
Source code in ccproxy/auth/storage/keyring.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
async def exists(self) -> bool:
    """Check if credentials exist in the keyring.

    Returns:
        True if credentials exist, False otherwise
    """
    try:
        import keyring
    except ImportError:
        return False

    try:
        password = keyring.get_password(self.service_name, self.username)
        return password is not None
    except Exception:
        return False

delete async 

delete()

Delete credentials from the keyring.

Returns:

Type Description
bool

True if deleted successfully, False otherwise

Raises:

Type Description
CredentialsStorageError

If there's an error deleting from keyring
Source code in ccproxy/auth/storage/keyring.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
async def delete(self) -> bool:
    """Delete credentials from the keyring.

    Returns:
        True if deleted successfully, False otherwise

    Raises:
        CredentialsStorageError: If there's an error deleting from keyring
    """
    try:
        import keyring
    except ImportError as e:
        raise CredentialsStorageError(
            "keyring package is required for keyring storage. "
            "Install it with: pip install keyring"
        ) from e

    try:
        if await self.exists():
            keyring.delete_password(self.service_name, self.username)
            logger.debug(
                "credentials_delete_completed",
                source="keyring",
                service_name=self.service_name,
            )
            return True
        return False
    except Exception as e:
        raise CredentialsStorageError(
            f"Error deleting credentials from keyring: {e}"
        ) from e

get_location 

get_location()

Get the storage location description.

Returns:

Type Description
str

Description of the keyring storage location
Source code in ccproxy/auth/storage/keyring.py 
186
187
188
189
190
191
192
def get_location(self) -> str:
    """Get the storage location description.

    Returns:
        Description of the keyring storage location
    """
    return f"OS keyring (service: {self.service_name}, user: {self.username})"