Implement more methods and fix security hotspot

.coveragerc

@@ -1,8 +1,6 @@
 [run]
 relative_files = True
-
-[report]
 include =
     src/*
 omit =
-    */tests/*
+    src/pcloud/tests/*

.github/workflows/pcloud-test.yml

@@ -11,11 +11,10 @@ on:
 
 jobs:
   build:
-
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.7, 3.8, 3.9, "3.10", "3.11"]
+        python-version: [3.7, 3.8, 3.9, "3.10", "3.11", "3.12"]
       max-parallel: 1
     steps:
     - uses: actions/checkout@v3
@@ -58,5 +57,6 @@ jobs:
           -Dsonar.organization=tomgross-github
           -Dsonar.python.version=3
           -Dsonar.python.coverage.reportPaths=coverage.xml
+          -Dsonar.python.coverage.exclusions=**/tests/*
 
 

.gitignore

@@ -40,4 +40,5 @@ pyvenv.cfg
 Pipfile*
 *.whl
 .vscode
-coverage.xml
+coverage.xml
+.env

README.rst

@@ -11,7 +11,6 @@ Features
 ========
 
 - Can be used as a library
-- Comes with a command line script
 - Provides a PyFileSystem implementation
 
 Examples

setup.py

@@ -32,6 +32,8 @@
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
         "Operating System :: OS Independent",
         "License :: OSI Approved :: MIT License",
         "Topic :: Software Development :: Libraries :: Python Modules",
@@ -48,13 +50,10 @@
     install_requires=[
         "requests",
         "requests-toolbelt",
-        "setuptools"
+        "setuptools>=68.0.0"
     ],
     extras_require={"pyfs": ["fs"]},
     entry_points={
-        "console_scripts": [
-            "pcloud-cli = pcloud.api:main",
-        ],
         "fs.opener": ["pcloud  = pcloud.pcloudfs:PCloudOpener"],
     },
 )

src/pcloud/api.py

@@ -7,7 +7,6 @@
 from urllib.parse import urlparse
 from urllib.parse import urlunsplit
 
-import argparse
 import datetime
 import logging
 import os.path
@@ -44,9 +43,11 @@ class OnlyPcloudError(NotImplementedError):
     """Feature restricted to pCloud"""
 
 
-# Helpers
+class InvalidFileModeError(Exception):
+    """File mode not supported"""
 
 
+# Helpers
 def to_api_datetime(dt):
     """Converter to a datetime structure the pCloud API understands
 
@@ -57,19 +58,6 @@ def to_api_datetime(dt):
     return dt
 
 
-def main():
-    parser = argparse.ArgumentParser(description="pCloud command line client")
-    parser.add_argument(
-        "username", help="The username for login into your pCloud account"
-    )
-    parser.add_argument(
-        "password", help="The password for login into your pCloud account"
-    )
-    args = parser.parse_args()
-    pyc = PyCloud(args.username, args.password)
-    print(pyc)
-
-
 class PyCloud(object):
     endpoints = {
         "api": "https://api.pcloud.com/",
@@ -251,13 +239,17 @@ def _upload(self, method, files, **kwargs):
             kwargs["auth"] = self.auth_token
         elif self.access_token:  # OAuth2 authentication
             kwargs["access_token"] = self.access_token
-        kwargs.pop("fd", None)
         fields = list(kwargs.items())
         fields.extend(files)
         m = MultipartEncoder(fields=fields)
-        resp = self.session.post(
+        # use own request and not session to make sure connection is closed after
+        # write
+        resp = requests.post(
             self.endpoint + method, data=m, headers={"Content-Type": m.content_type}
         )
+        import pdb; pdb.set_trace()
+        # fix lazy loading of response
+        # str(resp.content)
         return resp.json()
 
     @RequiredParameterCheck(("files", "data"))
@@ -380,7 +372,7 @@ def gettextfile(self, **kwargs):
     def file_open(self, **kwargs):
         return self._do_request("file_open", **kwargs)
 
-    @RequiredParameterCheck(("fd",))
+    @RequiredParameterCheck(("fd", "count"))
     def file_read(self, **kwargs):
         return self._do_request("file_read", json=False, **kwargs)
 
@@ -403,7 +395,9 @@ def file_truncate(self, **kwargs):
     @RequiredParameterCheck(("fd", "data"))
     def file_write(self, **kwargs):
         files = [("file", ("upload-file.io", BytesIO(kwargs.pop("data"))))]
+        kwargs["fd"] = str(kwargs["fd"])
         return self._upload("file_write", files, **kwargs)
+        # return self._do_request("file_write", **kwargs)
 
     @RequiredParameterCheck(("fd",))
     def file_pwrite(self, **kwargs):
@@ -462,6 +456,38 @@ def listshares(self, **kwargs):
         return self._do_request("listshares", **kwargs)
 
     # Public links
+    def getfilepublink(self, **kwargs):
+        raise OnlyPcloudError(ONLY_PCLOUD_MSG)
+
+    def getpublinkdownload(self, **kwargs):
+        raise OnlyPcloudError(ONLY_PCLOUD_MSG)
+
+    @RequiredParameterCheck(("path", "folderid"))
+    def gettreepublink(self, **kwargs):
+        raise NotImplementedError
+
+    @RequiredParameterCheck(("code",))
+    def showpublink(self, **kwargs):
+        return self._do_request("showpublink", authenticate=False, **kwargs)
+
+    @RequiredParameterCheck(("code",))
+    def copypubfile(self, **kwargs):
+        return self._do_request("copypubfile", **kwargs)
+
+    def listpublinks(self, **kwargs):
+        return self._do_request("listpublinks", **kwargs)
+
+    def listplshort(self, **kwargs):
+        return self._do_request("listplshort", **kwargs)
+
+    @RequiredParameterCheck(("linkid",))
+    def deletepublink(self, **kwargs):
+        return self._do_request("deletepublink", **kwargs)
+
+    @RequiredParameterCheck(("linkid",))
+    def changepublink(self, **kwargs):
+        return self._do_request("changepublink", **kwargs)
+
     @RequiredParameterCheck(("path", "folderid"))
     def getfolderpublink(self, **kwargs):
         expire = kwargs.get("expire")
@@ -509,6 +535,19 @@ def trash_restorepath(self, **kwargs):
     def trash_restore(self, **kwargs):
         raise NotImplementedError
 
+    # convenience methods
+    @RequiredParameterCheck(("path",))
+    def file_exists(self, **kwargs):
+        path = kwargs["path"]
+        resp = self.file_open(path=path, flags=O_APPEND)
+        result = resp.get("result")
+        if result == 0:
+            self.file_close(fd=resp["fd"])
+            return True
+        elif result == 2009:
+            return False
+        else:
+            raise OSError(f"pCloud error occured ({result}) - {resp['error']}:  {path}")
+
 
-if __name__ == "__main__":
-    main()
+# EOF

src/pcloud/oauth2.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 import _thread
+import time
 
 from http.server import BaseHTTPRequestHandler
 from http.server import HTTPServer
@@ -30,6 +31,13 @@ def do_GET(self):
             self.wfile.write(b"<html><h1>You may now close this window.</h1></html>")
 
 
+class HTTPServerWithAttributes(HTTPServer):
+    def __init__(self, *args, **kwargs):
+        self.access_token = None
+        self.pc_hostname = None
+        super().__init__(*args, **kwargs)
+
+
 class TokenHandler(object):
     """
     Class used to handle pClouds oAuth2
@@ -49,17 +57,15 @@ def close_browser(self):
         """Hook which is called after request is handled."""
 
     def get_access_token(self):
-        http_server = HTTPServer(("localhost", PORT), HTTPServerHandler)
+        http_server = HTTPServerWithAttributes(("localhost", PORT), HTTPServerHandler)
 
-        # Solution taken from https://stackoverflow.com/a/12651298
-        # There might be better ways than accessing the internal
-        # _thread library for starting the http-server non-blocking
-        # but I did not found any ;-)
         def start_server():
             http_server.handle_request()
 
         _thread.start_new_thread(start_server, ())
         self.open_browser()
+        while not (http_server.access_token and http_server.pc_hostname):
+            time.sleep(1)
         self.close_browser()
         http_server.server_close()
         return http_server.access_token, http_server.pc_hostname