diff --git a/src/pcloud/api.py b/src/pcloud/api.py index ee33c04..5e2d260 100644 --- a/src/pcloud/api.py +++ b/src/pcloud/api.py @@ -5,6 +5,12 @@ from hashlib import sha1 from io import BytesIO +from pcloud.protocols import JsonAPIProtocol +from pcloud.protocols import JsonEAPIProtocol +from pcloud.protocols import BinAPIProtocol +from pcloud.protocols import BinEAPIProtocol +from pcloud.protocols import TestProtocol +from pcloud.protocols import NearestProtocol from pcloud.jsonprotocol import PCloudJSONConnection from pcloud.oauth2 import TokenHandler from pcloud.utils import log @@ -41,12 +47,12 @@ class InvalidFileModeError(Exception): class PyCloud(object): endpoints = { - "api": "https://api.pcloud.com/", - "eapi": "https://eapi.pcloud.com/", - "test": "localhost:5023", - "binapi": "https://binapi.pcloud.com", - "bineapi": "https://bineapi.pcloud.com", - "nearest": "", + "api": JsonAPIProtocol, + "eapi": JsonEAPIProtocol, + "test": TestProtocol, + "binapi": BinAPIProtocol, + "bineapi": BinEAPIProtocol, + "nearest": NearestProtocol, } def __init__( @@ -55,8 +61,7 @@ def __init__( password, endpoint="api", token_expire=31536000, - oauth2=False, - connection=PCloudJSONConnection, + oauth2=False ): if endpoint not in self.endpoints: log.error( @@ -67,17 +72,11 @@ def __init__( return elif endpoint == "nearest": self.endpoint = self.getnearestendpoint() - elif endpoint not in connection.allowed_endpoints: - log.error( - "Endpoint (%s) not in allowed list of '%s'. Use one of: %s", - endpoint, - connection.__name__, - ", ".join(connection.allowed_endpoints), - ) - return + conn = PCloudJSONConnection(self) else: - self.endpoint = self.endpoints.get(endpoint) - conn = connection(self) + protocol = self.endpoints.get(endpoint) + self.endpoint = protocol.endpoint + conn = protocol.connection(self) self.connection = conn.connect() log.info(f"Using pCloud API endpoint: {self.endpoint}") @@ -107,15 +106,15 @@ def oauth2_authorize( See https://docs.pcloud.com/methods/oauth_2.0/authorize.html Per default the Python webbrowser library, which opens - a reals browser is used for URL redirection. + a real browser used for URL redirection. You can provide your own token handler (i.e. headless selenium), if needed. """ - ep = {urlparse(y).netloc: x for x, y in PyCloud.endpoints.items()} + ep = {urlparse(protocol.endpoint).netloc: key for key, protocol in PyCloud.endpoints.items()} code, hostname = tokenhandler(client_id).get_access_token() params = {"client_id": client_id, "client_secret": client_secret, "code": code} endpoint = ep.get(hostname) - endpoint_url = PyCloud.endpoints.get(endpoint) + endpoint_url = PyCloud.endpoints.get(endpoint).endpoint resp = requests.get(endpoint_url + "oauth2_token", params=params).json() access_token = resp.get("access_token") return cls("", access_token, endpoint, token_expire, oauth2=True) diff --git a/src/pcloud/binaryprotocol.py b/src/pcloud/binaryprotocol.py index 985458d..04cb951 100644 --- a/src/pcloud/binaryprotocol.py +++ b/src/pcloud/binaryprotocol.py @@ -21,7 +21,7 @@ class PCloudBinaryConnection(object): NOTE: .connect() must be called to establish network communication. """ - allowed_endpoints = frozenset(["binapi", "bineapi", "test", "nearest"]) + allowed_endpoints = frozenset(["binapi", "bineapi"]) def __init__(self, api, persistent_params=None): """Initializes the binary API. diff --git a/src/pcloud/dummyprotocol.py b/src/pcloud/dummyprotocol.py new file mode 100644 index 0000000..c625f6d --- /dev/null +++ b/src/pcloud/dummyprotocol.py @@ -0,0 +1,69 @@ +import requests + +from pcloud.utils import log +from requests_toolbelt.multipart.encoder import MultipartEncoder +from pcloud.jsonprotocol import PCloudJSONConnection + +class NoOpSession(object): + kwargs = {} + + def get(self, url, **kwargs): + self.kwargs = kwargs + self.kwargs["url"] = url + return self + + def json(self): + return self.kwargs + +class PCloudDummyConnection(PCloudJSONConnection): + """Connection to pcloud.com based on their JSON protocol.""" + + allowed_endpoints = frozenset(["test"]) + + def __init__(self, api): + """Connect to pcloud API based on their JSON protocol.""" + self.session = NoOpSession() + self.api = api + + def connect(self): + return self + + def do_get_request(self, method, authenticate=True, json=True, endpoint=None, **kw): + if "noop" in kw: + kw.pop("noop") + params = { + "params": kw, + "url": self.api.endpoint + method + } + return params + else: + return super().do_get_request(method, authenticate, json, endpoint, **kw) + + def upload(self, method, files, **kwargs): + if self.api.auth_token: # Password authentication + kwargs["auth"] = self.api.auth_token + elif self.api.access_token: # OAuth2 authentication + kwargs["access_token"] = self.api.access_token + fields = list(kwargs.items()) + fields.extend(files) + + # from requests import Request, Session + + # s = Session() + + # for entry in files: + # filename, fd = entry[1] + # fields["filename"] = filename + # req = Request('PUT', self.api.endpoint + method, data=fields) + # prepped = req.prepare() + # prepped.body = fd + # resp = s.send(prepped) + + # resp = self.session.post(self.api.endpoint + method, files=files, data=kwargs) + m = MultipartEncoder(fields=fields) + resp = requests.post( + self.api.endpoint + method, data=m, headers={"Content-Type": m.content_type} + ) + # data = dump_all(resp) + # print(data.decode('utf-8')) + return resp.json() diff --git a/src/pcloud/jsonprotocol.py b/src/pcloud/jsonprotocol.py index 156e61e..4cd1d5f 100644 --- a/src/pcloud/jsonprotocol.py +++ b/src/pcloud/jsonprotocol.py @@ -7,7 +7,7 @@ class PCloudJSONConnection(object): """Connection to pcloud.com based on their JSON protocol.""" - allowed_endpoints = frozenset(["api", "eapi", "test", "nearest"]) + allowed_endpoints = frozenset(["api", "eapi", "nearest"]) def __init__(self, api): """Connect to pcloud API based on their JSON protocol.""" diff --git a/src/pcloud/protocols.py b/src/pcloud/protocols.py new file mode 100644 index 0000000..6a47e9b --- /dev/null +++ b/src/pcloud/protocols.py @@ -0,0 +1,34 @@ +from pcloud.dummyprotocol import PCloudDummyConnection +from pcloud.jsonprotocol import PCloudJSONConnection +from pcloud.binaryprotocol import PCloudBinaryConnection + + +class TestProtocol(object): + name = "test" + endpoint = "http://localhost:5023/" + connection = PCloudDummyConnection + +class JsonAPIProtocol(object): + name = "api" + endpoint = "https://api.pcloud.com/" + connection = PCloudJSONConnection + +class JsonEAPIProtocol(object): + name = "eapi" + endpoint = "https://eapi.pcloud.com/" + connection = PCloudJSONConnection + +class BinAPIProtocol(object): + name = "binapi" + endpoint = "https://binapi.pcloud.com/" + connection = PCloudBinaryConnection + +class BinEAPIProtocol(object): + name = "bineapi" + endpoint = "https://bineapi.pcloud.com/" + connection = PCloudBinaryConnection + +class NearestProtocol(object): + name = "nearest" + endpoint = "" + connection = PCloudJSONConnection \ No newline at end of file diff --git a/src/pcloud/tests/test_api.py b/src/pcloud/tests/test_api.py index ff569fd..6eca82f 100644 --- a/src/pcloud/tests/test_api.py +++ b/src/pcloud/tests/test_api.py @@ -6,6 +6,7 @@ import json import os.path import pytest +import requests class NoOpSession(object): @@ -28,24 +29,31 @@ def get_auth_token(self): self.auth_token = None self.access_token = None else: - return super(DummyPyCloud, self).get_auth_token() + return super().get_auth_token() def __init__(self, username, password, noop=False): if noop: self.noop = True - super(DummyPyCloud, self).__init__(username, password, endpoint="test") + super().__init__(username, password, endpoint="test") if noop: self.session = NoOpSession() + def _do_request(self, method, authenticate=True, json=True, endpoint=None, **kw): + if self.noop: + kw["noop"] = True + return self.connection.do_get_request( + method, authenticate, json, endpoint, **kw + ) + class DummyPCloudFS(PCloudFS): factory = DummyPyCloud def test_getfolderpublink(): - api = DummyPyCloud("john", "doe", noop=True) + pcapi = DummyPyCloud("john", "doe", noop=True) dt = datetime.datetime(2023, 10, 5, 12, 3, 12) - assert api.getfolderpublink(folderid=20, expire=dt) == { + assert pcapi.getfolderpublink(folderid=20, expire=dt) == { "params": {"expire": "2023-10-05T12:03:12", "folderid": 20}, "url": "http://localhost:5023/getfolderpublink", } @@ -114,8 +122,8 @@ def test_getpublinkdownload(self): papi.getpublinkdownload(file=self.noop_dummy_file) def test_server_security(self): - api = DummyPyCloud("", "") - resp = api.session.get(api.endpoint + "../../bogus.sh", params={}) + papi = DummyPyCloud("", "") + resp = requests.get(papi.endpoint + "../../bogus.sh", params={}) assert resp.content == b'{"Error": "Path not found or not accessible!"}' assert resp.status_code == 404 @@ -138,9 +146,3 @@ def test_getpublinkdownload(self): papi = DummyPyCloud("foo", "bar") with pytest.raises(api.OnlyPcloudError): papi.getpublinkdownload(file=self.noop_dummy_file) - - def test_server_security(self): - api = DummyPyCloud("", "") - resp = api.session.get(api.endpoint + "../../bogus.sh", params={}) - assert resp.content == b'{"Error": "Path not found or not accessible!"}' - assert resp.status_code == 404 diff --git a/src/pcloud/tests/test_bin_integration.py b/src/pcloud/tests/test_bin_integration.py index be297f8..8a94055 100644 --- a/src/pcloud/tests/test_bin_integration.py +++ b/src/pcloud/tests/test_bin_integration.py @@ -17,8 +17,7 @@ def pycloud(): username = os.environ.get("PCLOUD_USERNAME") password = os.environ.get("PCLOUD_PASSWORD") return PyCloud( - username, password, endpoint="bineapi", connection=PCloudBinaryConnection - ) + username, password, endpoint="bineapi") folder_for_tests = "integration-bin-test"