Code Refactoring

mojang/__init__.py

@@ -1,3 +1,2 @@
-from .auth.session import user
-from .user.profile import profile
-from .user.api import *
+from .main import user
+from .api import names, status, uuid, uuids

mojang/api/__init__.py

@@ -0,0 +1 @@
+from .base import status, names, uuid, uuids

mojang/api/auth/yggdrasil.py

@@ -0,0 +1,81 @@
+import requests
+from ..urls import AUTHENTICATE, REFRESH, VALIDATE, SIGNOUT, INVALIDATE
+from ...error.handler import handle_response
+from ...error.exceptions import *
+
+def authenticate(session: requests.Session, username: str, password: str, client_token=None):
+    payload = {
+        'username': username,
+        'password': password,
+        'clientToken': client_token,
+        'agent': {
+            'name': 'Minecraft',
+            'version': 1
+        }
+    }
+
+    response = session.post(AUTHENTICATE, json=payload)
+    data = handle_response(response, PayloadError, CredentialsError)
+    session.headers.update({'Authorization': 'Bearer {}'.format(data['accessToken'])})
+
+    # Parse response
+    res = {
+        'access_token': data['accessToken'],
+        'client_token': data['clientToken'],
+        'uuid': data['selectedProfile']['id'],
+        'name': data['selectedProfile']['name'],
+        'legacy': data['selectedProfile'].get('legacy', False),
+        'demo': not data['selectedProfile'].get('paid', True)
+    }
+    return res
+
+def refresh(session: requests.Session, access_token: str, client_token: str):
+    payload = {
+        'accessToken': access_token,
+        'clientToken': client_token
+    }
+
+    response = session.post(REFRESH, json=payload)
+    data = handle_response(response, PayloadError, TokenError)
+    session.headers.update({'Authorization': 'Bearer {}'.format(data['accessToken'])})
+
+    # Parse response
+    res = {
+        'access_token': data['accessToken'],
+        'client_token': data['clientToken'],
+        'uuid': data['selectedProfile']['id'],
+        'name': data['selectedProfile']['name'],
+        'legacy': data['selectedProfile'].get('legacy', False),
+        'demo': not data['selectedProfile'].get('paid', True)
+    }
+    return res
+
+def validate(session: requests.Session, access_token: str, client_token: str):
+    payload = {
+        'accessToken': access_token,
+        'clientToken': client_token
+    }
+
+    response = session.post(VALIDATE, json=payload)
+    handle_response(response, PayloadError, TokenError)
+    return True
+
+def signout(session: requests.Session, username: str, password: str):
+    payload = {
+        'username': username,
+        'password': password
+    }
+
+    response = session.post(SIGNOUT, json=payload)
+    data = handle_response(response, PayloadError, CredentialsError)
+    session.headers.pop('Authorization')
+
+def invalidate(session: requests.Session, access_token: str, client_token: str):
+    payload = {
+        'accessToken': access_token,
+        'clientToken': client_token
+    }
+
+    response = session.post(INVALIDATE, json=payload)
+    handle_response(response, PayloadError, TokenError)
+    session.headers.pop('Authorization')

mojang/api/base.py

@@ -0,0 +1,50 @@
+import requests
+from .urls import STATUS_CHECK, NAME_HISTORY, GET_UUID, GET_UUIDS
+from ..error.exceptions import PayloadError
+from ..error.handler import handle_response
+
+
+def status(service=None):
+    res = {}
+    response = requests.get(STATUS_CHECK)
+    data = handle_response(response)
+    for status in data:
+        res.update(status)
+
+    if service:
+        return res[service]
+
+    return res
+
+def names(player_id: str):
+    response = requests.get(NAME_HISTORY.format(uuid=player_id))
+
+    names = []
+    data = handle_response(response)
+    for item in data:
+        if 'changedToAt' in item:
+            item['changedToAt'] = dt.datetime.fromtimestamp(item['changedToAt'])
+        names.append((item['name'], item.get('changedToAt',None)))
+
+    return names
+
+def uuid(username: str, only_uuid=True):
+    response = requests.get(GET_UUID.format(name=username))
+    data = handle_response(response)
+    data['uuid'] = data.pop('id')
+    data['legacy'] = data.get('legacy', False)
+    data['demo'] = data.get('demo', False)
+    if only_uuid:
+        return data['uuid']
+
+    return data
+
+def uuids(usernames: list, only_uuid=True):
+    data = {}
+    if len(usernames) > 0:
+        response = requests.post(GET_UUIDS, json=usernames)
+        data = handle_response(response, PayloadError)
+        if only_uuid:
+            data = list(map(lambda pdata: pdata['id'], data))
+
+    return data

mojang/api/security.py

@@ -0,0 +1,27 @@
+from .urls import SECURITY_CHECK, SECURITY_CHALLENGES
+from ..error.handler import handle_response
+from ..error.exceptions import *
+
+
+def is_secure(session):
+    response = session.get(SECURITY_CHECK)
+    try:
+        handle_response(response, PayloadError, Unauthorized, IPNotSecured)
+    except IPNotSecured:
+        return False
+    else:
+        return True
+
+def get_challenges(session):
+    response = session.get(SECURITY_CHALLENGES)
+    handle_response(response, PayloadError, Unauthorized)
+    return data
+
+def verify_ip(session, answers: list):
+    response = session.post(SECURITY_CHECK, json=answers)
+    try:
+        handle_response(response, PayloadError, Unauthorized, IPVerificationError)
+    except IPVerificationError:
+        return False
+    else:
+        return True

mojang/api/urls.py

@@ -0,0 +1,36 @@
+from ..utils.url import URL
+
+# Base urls
+MOJANG_AUTHSERVER = URL('https://authserver.mojang.com')
+MOJANG_API = URL('https://api.mojang.com')
+MOJANG_STATUS = URL('https://status.mojang.com')
+MOJANG_SESSION = URL('https://sessionserver.mojang.com')
+MINECRAFT_SERVICES = URL('https://api.minecraftservices.com')
+
+# Status check
+STATUS_CHECK = MOJANG_STATUS.join('check')
+
+# Mojang authserver
+AUTHENTICATE = MOJANG_AUTHSERVER.join('authenticate')
+VALIDATE = MOJANG_AUTHSERVER.join('validate')
+REFRESH = MOJANG_AUTHSERVER.join('refresh')
+SIGNOUT = MOJANG_AUTHSERVER.join('signout')
+INVALIDATE = MOJANG_AUTHSERVER.join('invalidate')
+
+# Security Check
+SECURITY_CHECK = MOJANG_API.join('user/security/location')
+SECURITY_CHALLENGES = MOJANG_API.join('user/security/challenges')
+
+# Other
+NAME_HISTORY = MOJANG_API.join('user/profiles/{uuid}/names')
+GET_UUID = MOJANG_API.join('users/profiles/minecraft/{name}')
+GET_UUIDS = MOJANG_API.join('profiles/minecraft')
+
+CHECK_NAME_CHANGE = MINECRAFT_SERVICES.join('minecraft/profile/namechange')
+CHANGE_NAME = MINECRAFT_SERVICES.join('minecraft/profile/name/{name}')
+
+UPLOAD_SKIN = MINECRAFT_SERVICES.join('minecraft/profile/skins')
+RESET_SKIN = MOJANG_API.join('user/profile/{uuid}/skin')
+
+GET_PROFILE = MOJANG_SESSION.join('session/minecraft/profile/{uuid}')
+GET_AUTH_PROFILE = MINECRAFT_SERVICES.join('minecraft/profile')

mojang/api/user.py

@@ -0,0 +1,83 @@
+import datetime as dt
+import json
+from base64 import urlsafe_b64decode
+
+import requests
+
+from ..error.exceptions import *
+from ..error.handler import handle_response
+from ..utils.cape import Cape
+from ..utils.skin import Skin
+from .urls import (CHANGE_NAME, CHECK_NAME_CHANGE, GET_AUTH_PROFILE, GET_PROFILE, RESET_SKIN, UPLOAD_SKIN)
+
+
+def check_name_change(session: requests.Session):
+    res = dict.fromkeys(('created_at','name_change_allowed'), None)
+
+    if 'Authorization' in session.headers.keys():
+        response = session.get(CHECK_NAME_CHANGE)
+        data = handle_response(response, PayloadError, Unauthorized)
+        res.update({
+            'created_at': dt.datetime.strptime(data['createdAt'], '%Y-%m-%dT%H:%M:%SZ'),
+            'name_change_allowed': data['nameChangeAllowed']
+        })
+
+    return res
+
+def change_name(session: requests.Session, name: str):
+    response = session.put(CHANGE_NAME)
+    handle_response(response, InvalidName, UnavailableName, Unauthorized)
+
+def upload_skin(session: requests.Session, path: str, variant='classic'):
+    skin = Skin(path, variant=variant)
+    skin_data = skin.data
+    files = [
+        ('variant', variant),
+        ('file', (f'image.{skin.extension[1:]}', skin_data, f'image/{skin.extension[1:]}'))
+    ]
+
+    response = session.post(UPLOAD_SKIN, files=files, headers={'Content-Type': None})
+    handle_response(response, PayloadError, Unauthorized)
+
+def reset_skin(session: requests.Session):
+    response = session.delete(RESET_SKIN)
+    handle_response(response, PayloadError, Unauthorized)
+
+def get_profile(session: requests.Session, uuid=None):
+    res = dict.fromkeys(('uuid','name','skins','capes'), None)
+
+    if 'Authorization' in session.headers.keys():
+        response = session.get(GET_AUTH_PROFILE)
+        data = handle_response(response, PayloadError, Unauthorized)
+
+        res.update({
+            'uuid': data['id'],
+            'name': data['name'],
+            'skins': [],
+            'capes': []
+        })
+
+        for skin in data['skins']:
+            res['skins'].append(Skin(skin['url'], skin['variant'].lower()))
+
+        for cape in data['capes']:
+            res['skins'].append(Cape(cape['url']))
+    else:
+        response = session.get(GET_PROFILE.format(uuid=uuid))
+        data = handle_response(response, PayloadError)
+
+        res.update({
+            'uuid': data['id'],
+            'name': data['name'],
+            'skins': [],
+            'capes': []
+        })
+
+        for d in data['properties']:
+            textures = json.loads(urlsafe_b64decode(d['value']))['textures']
+            if 'SKIN' in textures.keys():
+                res['skins'].append(Skin(textures['SKIN']['url'], textures['SKIN'].get('metadata',{}).get('model','classic')))
+            if 'CAPE' in textures.keys():
+                res['skins'].append(Cape(textures['CAPE']['url']))
+
+    return res