-
Notifications
You must be signed in to change notification settings - Fork 0
CDNClient
steam.client.cdn.CDNClient is Steam's content-depot client. It talks to SteamPipe (the CDN behind actual game files), fetches encrypted manifests, and reads decrypted chunks on demand. Behaves like a lazy filesystem over the CDN — you can iterate every file in a depot, or open a single file and .read() it byte-for-byte, without downloading the rest.
- The
clientextra (gevent,protobuf,gevent-eventemitter). - A logged-in
SteamClientinstance. Anonymous logins won't have licenses for anything, so you can only fetch depots that are publicly available. See FAQ for the "anonymous vs. credentialled" tradeoff.
CDNClient uses requests under the hood, which isn't gevent-cooperative by default. The recommended pattern:
import steam.monkey
steam.monkey.patch_minimal()
from steam.client import SteamClient
from steam.client.cdn import CDNClientWithout the patch, HTTP calls block the gevent loop.
CDNClient(client)client is a logged-in SteamClient. On construction, CDNClient:
- Bootstraps its cell id from the SteamClient.
- Fetches a list of content servers from
IContentServerDirectoryService.GetServersForSteamPipe. - Loads licenses from the SteamClient — populates
licensed_app_idsandlicensed_depot_ids, which gate every subsequent request.
mysteam = SteamClient()
mysteam.cli_login()
mycdn = CDNClient(mysteam)Anonymous logins get exactly one package (17906 — the "Steam free products" package). Any other content raises a SteamError because you have no license.
get_manifests(app_id, branch='public', password=None, filter_func=None, decrypt=True)Returns a list of CDNDepotManifest, one per depot the app publishes on that branch that you have a license for. Encrypted (beta) branches take a password.
>>> mycdn.get_manifests(570)
[<CDNDepotManifest('Dota 2 Content', app_id=570, depot_id=373301, gid=6397590570861788404, creation_time='2019-06-29 16:03:11')>,
<CDNDepotManifest('Dota 2 Content 2', app_id=570, depot_id=381451, ...)>,
...]Filter with a callback that receives the depot id + the parsed depot info dict from PICS:
mycdn.get_manifests(570, filter_func=lambda depot_id, info: 'Content' in info.get('name', ''))get_manifest(app_id, depot_id, manifest_gid, decrypt=True, manifest_request_code=0)Lower-level entry point when you already know the depot + manifest ids:
mymanifest = mycdn.get_manifest(570, 373301, 6397590570861788404)get_manifests() fetches the request code for you; get_manifest() accepts one directly for callers that already hold one (via get_manifest_request_code).
iter_files(app_id, filename_filter=None, branch='public', password=None, filter_func=None)Generator over CDNDepotFile across every manifest of the app. filename_filter is a shell-wildcard pattern:
for f in mycdn.iter_files(570, r'game\dota\gameinfo.gi'):
print(f)
print(f.read(80).decode('utf-8'))Output:
<CDNDepotFile(570, 373301, 6397590570861788404, 'game\\dota\\gameinfo.gi', 6808)>
"GameInfo"
{
game "Dota 2"
title "Dota 2"
...
CDNDepotFile implements the file-like protocol: .read(length=-1), .readline(), .readlines(), .seek(offset, whence=0), .tell(). Iterating over it yields lines. Files are .seekable if they're not directories or symlinks.
For fine-grained control (streaming into a decoder, partial downloads):
get_chunk(app_id, depot_id, chunk_id)Returns the decrypted (and, if applicable, LZMA- or zip-decompressed) chunk bytes. chunk_id is the SHA-1 of the chunk in hex. The chunk cache is a cachetools.LRUCache(20) — repeated reads of the same chunk are free.
Under the hood, .read() figures out which chunks intersect the requested (offset, length) slice and pulls just those.
get_depot_key(app_id, depot_id)Returns the AES key needed to decrypt manifests and chunks for a given depot. Cached in mycdn.depot_keys. Calls out to SteamClient.get_depot_key internally — needs a live CM login.
get_manifest_for_workshop_item(item_id)Given a published-file id, return the underlying CDNDepotManifest for that workshop item. Useful when you want to iterate a mod's assets.
CDNClient caches PICS lookups it makes:
-
mycdn.app_depots[app_id]— depot info dict fromclient.get_product_info([app_id])['apps'][app_id]['depots']. -
mycdn.manifests[(app_id, depot_id, manifest_gid)]— parsedCDNDepotManifestinstances. -
mycdn.depot_keys[depot_id]— AES depot keys. -
mycdn.beta_passwords[(app_id, branch)]— decrypted branch passwords.
mycdn.clear_cache() wipes the first two.
check_beta_password(app_id, password)Unlocks encrypted branches. Once unlocked, get_manifests(app_id, branch='dev', password='...') picks up the depot key transparently.
has_license_for_depot(depot_id) — quick check whether the SteamClient's account has access to the depot. Used internally by get_manifests to skip depots you can't fetch anyway (would fail on the depot-key request otherwise).
Anonymous logins have licensed_app_ids == the free products package, and nothing else. If you need real game files, log in with credentials.
-
SteamError— most CDN failures. Carries anEResultin.eresultwhere applicable. -
ManifestError— subclass ofSteamError. Attributes:.app_id,.depot_id,.manifest_gid,.error_msg,.error(inner exception).
- Need to sign in as a real user first? SteamClient covers login.
- Want to explore what a depot contains before downloading? Use
iter_files(app_id, filename_filter=...)— it's lazy, cheap, and lets you dry-run the filename pattern. - Actually mirroring a depot to disk? Use the pattern
for f in iter_files(...): open(dest, 'wb').write(f.read())— files are seekable and small chunk reads are cache-friendly.
H47R15/steam — maintained fork of ValvePython/steam. MIT licensed.