"Command line Snapchat" or "Why Snapchat needs to fix its private API".

Snapchat.py

#!/usr/bin/python3
import sys
import os
import json
from time import (time, sleep)
from hashlib import sha256
from zipfile import ZipFile
from io import (BytesIO, StringIO)

try :
    from Crypto.Cipher import AES
    _CAN_DECRYPT = True
except ImportError :
    _CAN_DECRYPT = False

try :
    from moviepy.editor import (VideoFileClip, ImageClip, CompositeVideoClip)
    _CAN_SAVE_ZIPPED = True
except ImportError :
    _CAN_SAVE_ZIPPED = False

import requests

class SnapchatError(RuntimeError) :
    """Something Went Wrong with Snapchat"""

class Snapchat(object) :
    USER_AGENT = "Snapchat/4.1.01 (Nexus 4; Android 18)"
    API_URI    = "https://feelinsonice.appspot.com/"

    # Request Tokens
    STATIC_TOKEN  = "X"
    ENCRYPT_KEY   = "X"
    TOKEN_PATTERN = "X"
    TOKEN_SECRET  = "X"

    # Media Types
    IMAGE                        = 0
    VIDEO                        = 1
    VIDEO_NOAUDIO                = 2
    FRIEND_REQUEST               = 3
    FRIEND_REQUEST_IMAGE         = 4
    FRIEND_REQUEST_VIDEO         = 5
    FRIEND_REQUEST_VIDEO_NOAUDIO = 6

    # Media States
    NONE       = -1
    SENT       = 0
    DELIVERED  = 1
    VIEWED     = 2
    SCREENSHOT = 3

    # Snapchat Attributes
    username  = False
    data      = False
    token     = False
    silent    = False
    snappath  = False

    def __init__(self, snappath = False) :
        self.snappath = snappath if snappath else (os.path.dirname(os.path.realpath(__file__)) + "/snaps/")

        if not os.path.exists(self.snappath) :
            try :
                os.makedirs(self.snappath)

            except OSError as e :
                raise SnapchatError("Could not create the snapchat save directory \"{}\" - {}".format(self.snappath, e.strerror))

    def generateToken(self, timestamp, token = False) :
        token = token if token else self.token
        a = sha256((self.TOKEN_SECRET + token).encode("UTF-8")).hexdigest()
        b = sha256((str(timestamp) + self.TOKEN_SECRET).encode("UTF-8")).hexdigest()
        t = [a[i] if c == "0" else b[i] for i, c in enumerate(self.TOKEN_PATTERN)]
        return "".join(t)

    def generateStaticToken(self, timestamp) :
        return self.generateToken(timestamp, self.STATIC_TOKEN)

    def isValidSnap(self, s) :
      return (("rp" not in s) and
        ("broadcast" not in s) and
        (s["st"] in (self.SENT, self.DELIVERED)) and
        (s["m"] in (self.IMAGE, self.VIDEO, self.VIDEO_NOAUDIO)))

    def apicall(self, module, data, stream = False) :
        uri = self.API_URI + module
        headers = {"user-agent" : self.USER_AGENT}
        return requests.post(uri, data=data, headers=headers, stream=stream)

    def register(self, username, email, password, age, dob) :
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateStaticToken(timestamp),
            "email"     : email,
            "password"  : password,
            "age"       : age,
            "birthday"  : dob
        }

        res = self.apicall("bq/register", req)
        if res.status_code == requests.codes.ok :
            ret = json.loads(res.content.decode("UTF-8"))
            if ret["logged"] == True :
                    self.token = ret["token"]
                    self.attachUser(email, username)
                    return True
            else:
                raise SnapchatError("Could not register user - {}".format(ret["message"]))
        else :
            raise SnapchatError("Could not register user - apicall status code {}".format(res.status_code))

    def attach(self, email, username) :
        """Attaches a username to an account"""
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateStaticToken(timestamp),
            "email"     : email,
            "username"  : username
        }

        res = self.apicall("ph/registeru", req)
        if res.status_code == requests.codes.ok :
            ret = json.loads(res.content.decode("UTF-8"))
            if ret["logged"] == True :
                self.token = ret["auth_token"]
                self.data = data
                self.username = username
                return True
            else :
                raise SnapchatError("Could not attach username to account - {}".format(ret["message"]))
        else :
            raise SnapchatError("Could not attach username to account - apicall status code {}".format(res.status_code))

    def login(self, username, password) :
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateStaticToken(timestamp),
            "username"  : username,
            "password"  : password
        }

        res = self.apicall("bq/login", req)
        if res.status_code == requests.codes.ok :
            ret = json.loads(res.content.decode("UTF-8"))
            if ret["logged"] == True :
                self.token = ret["auth_token"]
                self.data = ret
                self.username = username
                return True
            else :
                raise SnapchatError("Could not login - {}".format(ret["message"]))
        else :
            raise SnapchatError("Could not login - apicall status code {}".format(res.status_code))

    def logout(self) :
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateToken(timestamp),
            "username"  : self.username,
            "json"      : "{}",
            "events"    : "[]"
        }

        res = self.apicall("ph/logout", req)
        if res.status_code == requests.codes.ok :
            self.data      = False
            self.token     = False
            self.username  = False
            return True
        else :
            raise SnapchatError("Could not logout - apicall status code {}".format(res.status_code))

    def update(self) :
        """Pulls Snapchat updates"""
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateToken(timestamp),
            "username"  : self.username
        }

        res = self.apicall("bq/updates", req)
        if res.status_code == requests.codes.ok :
            ret = json.loads(res.content.decode("UTF-8"))
            if ret["logged"] == True :
                self.data = ret
                return True
            else :
                raise SnapchatError("Could not pull updates - {}".format(ret["message"]))
        else :
            raise SnapchatError("Could not pull updates - apicall status code {}".format(res.status_code))

    def snapcount(self) :
        return len([s for s in self.data["snaps"] if self.isValidSnap(s)])

    def clear(self) :
        """Clears the Snapchat queue of Snaps"""
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateToken(timestamp),
            "username"  : self.username,
        }

        res = self.apicall("ph/clear", req)

        if res.status_code == requests.codes.ok :
            return True
        else :
            raise SnapchatError("Could not clear feed - apicall status code {}".format(res.status_code))

    def getSnaps(self) :
        """Fetches all Snaps"""
        for s in self.data["snaps"] :
            if (self.isValidSnap(s)):
              self.getSnap(s)

    def getSnap(self, snap) :
        """Fetches a snap, decrypts it and saves it (if possible)"""
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateToken(timestamp),
            "username"  : self.username,
            "id"        : snap["id"]
        }

        res = self.apicall("ph/blob", req, True)
        if res.status_code == requests.codes.ok :
            try :
                if _CAN_DECRYPT :
                    data = res.raw.read()

                    unen = AES.new(self.ENCRYPT_KEY, AES.MODE_ECB)
                    data = unen.decrypt(data)

                    if snap["m"] in (self.IMAGE, self.VIDEO, self.VIDEO_NOAUDIO) :
                        saved = False
                        if "zipped" in snap :
                            if _CAN_SAVE_ZIPPED :
                                os.makedirs("{}temp/".format(self.snappath))
                                with ZipFile(BytesIO(data)) as z :
                                    for m in z.infolist() :
                                        part = m.filename.split("~")
                                        f = z.read(m)
                                        if part[0] == "media" :
                                            ext = "media.mp4"
                                        elif part[0] == "overlay" :
                                            ext = "overlay.png"
                                        open("{}temp/{}".format(self.snappath, ext), "wb").write(f)

                                media = VideoFileClip("{}temp/media.mp4".format(self.snappath))
                                overlay = ImageClip("{}temp/overlay.png".format(self.snappath)).resize(0.5)
                                clip = CompositeVideoClip([media, overlay]).\
                                set_duration(media.duration).\
                                to_videofile("{}{}_{}.mp4".format(self.snappath, self.username, snap["id"]),
                                             verbose=False,
                                             codec="mpeg4")
                                os.remove("{}temp/media.mp4".format(self.snappath))
                                os.remove("{}temp/overlay.png".format(self.snappath))
                                os.removedirs("{}temp/".format(self.snappath))
                                saved = True

                        else :
                            ext = "jpg" if snap["m"] == self.IMAGE else "mp4"
                            open("{}{}_{}.{}".format(self.snappath, self.username, snap["id"], ext), "wb").write(data)
                            saved = True

                        if not self.silent and saved :
                            self.markAsSeen(snap["id"])

            except Exception as e :
                raise SnapchatError("Could not save the snap - {}".format(e))

        elif res.status_code == 410 :
            raise SnapchatError("Could not fetch snap, either it doesn't exist or it's already been marked as viewed (Snap ID {} from {})".format(snap["id"], snap["sn"]))
        else :
            raise SnapchatError("Could not fetch snap - apicall status code {}".format(res.status_code))

    def markAsSeen(self, snapid) :
        """Notifies Snapchat that a Snap has been viewed"""
        timestamp = int(time())
        update = {
            snapid : {
                "c"        : 0,
                "t"        : timestamp,
                "replayed" : 0
            }
         }
        req = {
            "timestamp"               : timestamp,
            "req_token"               : self.generateToken(timestamp),
            "username"                : self.username,
            "added_friends_timestamp" : self.data["added_friends_timestamp"],
            "json"                    : json.dumps(update)
        }

        res = self.apicall("bq/update_snaps", req)

        if res.status_code == requests.codes.ok :
            return True
        else :
            raise SnapchatError("Could not mark snap as viewed - apicall status code {}".format(res.status_code))

    def setBirthday(self, birthday) :
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateToken(timestamp),
            "username"  : self.username,
            "action"    : "updateBirthday",
            "birthday"  : birthday
        }

        res = self.apicall("ph/settings", req)
        if res.status_code == requests.codes.ok :
            ret = json.loads(res.content.decode("UTF-8"))
            if ret["logged"] == True :
                return True
            else :
                raise SnapchatError("Could not update birthday - {}".format(ret["message"]))
        else :
            raise SnapchatError("Could not update birthday - apicall status code {}".format(res.status_code))

    def setEmail(self, email) :
        timestamp = int(time())
        req = {
            "timestamp" : timestamp,
            "req_token" : self.generateToken(timestamp),
            "username"  : self.username,
            "action"    : "updateEmail",
            "email"     : email
        }

        res = self.apicall("ph/settings", req)
        if res.status_code == requests.codes.ok :
            ret = json.loads(res.content.decode("UTF-8"))
            if ret["logged"] == True :
                return True
            else :
                raise SnapchatError("Could not update email address - {}".format(ret["message"]))
        else :
            raise SnapchatError("Could not update email address - apicall status code {}".format(res.status_code))

    def setPrivacy(self, setting) :
        """Sets account privacy (0 = snaps from anyone, 1 = friends only)"""
        timestamp = int(time())
        req = {
            "timestamp"      : timestamp,
            "req_token"      : self.generateToken(timestamp),
            "username"       : self.username,
            "action"         : "updatePrivacy",
            "privacySetting" : setting
        }

        res = self.apicall("ph/settings", req)
        if res.status_code == requests.codes.ok :
            ret = json.loads(res.content.decode("UTF-8"))
            if ret["logged"] == True :
                return True
            else :
                raise SnapchatError("Could not update privacy settings - {}".format(ret["message"]))
        else :
            raise SnapchatError("Could not attach update privacy settings - apicall status code {}".format(res.status_code))

if __name__ == '__main__' :
    import argparse
    import getpass

    p = argparse.ArgumentParser()
    p.add_argument("username")
    p.add_argument("-p", "--password", dest="password")
    p.add_argument("-s", "--silent", action="store_true", dest="silent")
    p.add_argument("-d", "--directory", dest="directory")
    p.add_argument("-v", "--verbose", action="store_true", dest="verbose")
    p = p.parse_args()

    def debug(message) :
        if p.verbose : print(message)

    debug("[\033[1;34m>>>\033[0m] \033[1mCommand line Snapchat\033[0m")
    debug("[\033[1;34m>>>\033[0m] {} decrypt Snaps".format("\033[1;32mCAN\033[0m" if _CAN_DECRYPT else "\033[1;31mCANNOT\033[0m"))
    debug("[\033[1;34m>>>\033[0m] {} save zipped Snaps".format("\033[1;32mCAN\033[0m" if _CAN_SAVE_ZIPPED else "\033[1;31mCANNOT\033[0m"))

    try :
        save = p.directory if p.directory else False
        snap = Snapchat(save)
        snap.silent = p.silent
        debug("[\033[1;34m>>>\033[0m] Signing in as {}..".format(p.username))
        password = p.password if p.password else getpass.getpass("[\033[1;35m***\033[0m] Password: ")
        snap.login(p.username, password)
        debug("[\033[1;34m<<<\033[0m] Signed in")

    except SnapchatError as e :
        debug("[\033[1;37;41m!!!\033[0m] {}".format(e))
        exit()


    while True :
        try :
            try :
                debug("[\033[1;34m>>>\033[0m] {}etching latest Snaps".format("Silently f" if p.silent else "F"))
                snap.update()
                num = snap.snapcount()
                debug("[\033[1;34m<<<\033[0m] {} Pending Snap{}".format(num, ("s" if num != 1 else "")))
                snap.getSnaps()

            except SnapchatError as e:
                debug("[\033[1;37;41m!!!\033[0m] {}".format(e))

            sleep(30)

        except KeyboardInterrupt :
            debug("[\033[1;34m>>>\033[0m] Exiting")
            exit()