Add FTP support to jobstores

docs/appendices/environment_vars.rst

@@ -217,6 +217,17 @@ There are several environment variables that affect the way Toil runs.
 |                                  | to S3 (``True`` by default).                       |
 |                                  | Example: ``TOIL_S3_USE_SSL=False``                 |
 +----------------------------------+----------------------------------------------------+
+| TOIL_FTP_USER                    | The FTP username to override all FTP logins with   |
+|                                  | Example: ``TOIL_FTP_USER=ftp_user``                |
++----------------------------------+----------------------------------------------------+
+| TOIL_FTP_PASSWORD                | The FTP password to override all FTP logins with   |
+|                                  | Example: ``TOIL_FTP_PASSWORD=ftp_password``        |
++----------------------------------+----------------------------------------------------+
+| TOIL_FTP_INSECURE                | Enable or disable connecting to all FTP            |
+|                                  | connections in insecure mode                       |
+|                                  | (``False`` by default).                            |
+|                                  | Example: ``TOIL_FTP_INSECURE=True``                |
++----------------------------------+----------------------------------------------------+
 | TOIL_WES_BROKER_URL              | An optional broker URL to use to communicate       |
 |                                  | between the WES server and Celery task queue. If   |
 |                                  | unset, ``amqp://guest:guest@localhost:5672//`` is  |

src/toil/jobStores/abstractJobStore.py

@@ -11,7 +11,9 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import ftplib
 import logging
+import netrc
 import os
 import pickle
 import re
@@ -35,7 +37,7 @@
 )
 from urllib.error import HTTPError
 from urllib.parse import ParseResult, urlparse
-from urllib.request import urlopen
+from urllib.request import urlopen, Request
 from uuid import uuid4
 
 from toil.common import Config, getNodeID, safeUnpickleFromStream
@@ -47,6 +49,7 @@
     ServiceJobDescription,
 )
 from toil.lib.compatibility import deprecated
+from toil.lib.conversions import strtobool
 from toil.lib.io import WriteWatchingStream
 from toil.lib.memoize import memoize
 from toil.lib.retry import ErrorCondition, retry
@@ -1870,15 +1873,28 @@ class JobStoreSupport(AbstractJobStore, metaclass=ABCMeta):
     stores.
     """
 
+    ftp = None
+
+    @classmethod
+    def _setup_ftp(cls) -> None:
+        if cls.ftp is None:
+            cls.ftp = FtpFsAccess(insecure=strtobool(os.environ.get('TOIL_FTP_INSECURE', 'False')) is True)
+
     @classmethod
     def _supports_url(cls, url: ParseResult, export: bool = False) -> bool:
         return url.scheme.lower() in ("http", "https", "ftp") and not export
 
     @classmethod
     def _url_exists(cls, url: ParseResult) -> bool:
+        # Deal with FTP first to support user/password auth
+        if url.scheme.lower() == "ftp":
+            cls._setup_ftp()
+            # mypy is unable to understand that ftp must exist by this point
+            assert cls.ftp is not None
+            return cls.ftp.exists(url.geturl())
+
         try:
-            # TODO: Figure out how to HEAD instead of this.
-            with cls._open_url(url):
+            with closing(urlopen(Request(url.geturl(), method="HEAD"))):
                 return True
         except FileNotFoundError:
             return False
@@ -1894,11 +1910,15 @@ def _url_exists(cls, url: ParseResult) -> bool:
     )
     def _get_size(cls, url: ParseResult) -> Optional[int]:
         if url.scheme.lower() == "ftp":
-            return None
-        with closing(urlopen(url.geturl())) as readable:
-            # just read the header for content length
-            size = readable.info().get("content-length")
-            return int(size) if size is not None else None
+            cls._setup_ftp()
+            # mypy is unable to understand that ftp must exist by this point
+            assert cls.ftp is not None
+            return cls.ftp.size(url.geturl())
+
+        # just read the header for content length
+        resp = urlopen(Request(url.geturl(), method="HEAD"))
+        size = resp.info().get("content-length")
+        return int(size) if size is not None else None
 
     @classmethod
     def _read_from_url(
@@ -1931,6 +1951,14 @@ def count(l: int) -> None:
         ]
     )
     def _open_url(cls, url: ParseResult) -> IO[bytes]:
+        # Deal with FTP first so we support user/password auth
+        if url.scheme.lower() == "ftp":
+            cls._setup_ftp()
+            # mypy is unable to understand that ftp must exist by this point
+            assert cls.ftp is not None
+            # we open in read mode as write mode is not supported
+            return cls.ftp.open(url.geturl(), mode="r")
+
         try:
             return cast(IO[bytes], closing(urlopen(url.geturl())))
         except HTTPError as e:
@@ -1958,3 +1986,142 @@ def _get_is_directory(cls, url: ParseResult) -> bool:
     def _list_url(cls, url: ParseResult) -> list[str]:
         # TODO: Implement HTTP index parsing and FTP directory listing
         raise NotImplementedError("HTTP and FTP URLs cannot yet be listed")
+
+
+class FtpFsAccess:
+    """
+    FTP access with upload.
+
+    Taken and modified from https://github.com/ohsu-comp-bio/cwl-tes/blob/03f0096f9fae8acd527687d3460a726e09190c3a/cwl_tes/ftp.py#L8
+    """
+
+    def __init__(
+        self, cache: Optional[dict[Any, ftplib.FTP]] = None, insecure: bool = False
+    ):
+        self.cache = cache or {}
+        self.netrc = None
+        self.insecure = insecure
+        try:
+            if "HOME" in os.environ:
+                if os.path.exists(os.path.join(os.environ["HOME"], ".netrc")):
+                    self.netrc = netrc.netrc(os.path.join(os.environ["HOME"], ".netrc"))
+            elif os.path.exists(os.path.join(os.curdir, ".netrc")):
+                self.netrc = netrc.netrc(os.path.join(os.curdir, ".netrc"))
+        except netrc.NetrcParseError as err:
+            logger.debug(err)
+
+    def exists(self, fn: str) -> bool:
+        return self.isfile(fn) or self.isdir(fn)
+
+    def isfile(self, fn: str) -> bool:
+        ftp = self._connect(fn)
+        if ftp:
+            try:
+                if not self.size(fn) is None:
+                    return True
+                else:
+                    return False
+            except ftplib.all_errors:
+                return False
+        return False
+
+    def isdir(self, fn: str) -> bool:
+        ftp = self._connect(fn)
+        if ftp:
+            try:
+                cwd = ftp.pwd()
+                ftp.cwd(urlparse(fn).path)
+                ftp.cwd(cwd)
+                return True
+            except ftplib.all_errors:
+                return False
+        return False
+
+    def open(self, fn: str, mode: str) -> IO[bytes]:
+        if 'r' in mode:
+            host, user, passwd, path = self._parse_url(fn)
+            handle = urlopen(
+                "ftp://{}:{}@{}/{}".format(user, passwd, host, path))
+            return cast(IO[bytes], closing(handle))
+        # TODO: support write mode
+        raise Exception('Write mode FTP not implemented')
+
+    def _parse_url(
+        self, url: str
+    ) -> tuple[Optional[str], Optional[str], Optional[str], str]:
+        parse = urlparse(url)
+        user = parse.username
+        passwd = parse.password
+        host = parse.hostname
+        path = parse.path
+        if parse.scheme == "ftp":
+            if not user and self.netrc:
+                if host is not None:
+                    creds = self.netrc.authenticators(host)
+                    if creds:
+                        user, _, passwd = creds
+        if not user:
+            if host is not None:
+                user, passwd = self._recall_credentials(host)
+                if passwd is None:
+                    passwd = "anonymous@"
+                    if user is None:
+                        user = "anonymous"
+
+        return host, user, passwd, path
+
+    def _connect(self, url: str) -> Optional[ftplib.FTP]:
+        parse = urlparse(url)
+        if parse.scheme == "ftp":
+            host, user, passwd, _ = self._parse_url(url)
+            if host is None:
+                # there has to be a host
+                return None
+            if (host, user, passwd) in self.cache:
+                if self.cache[(host, user, passwd)].pwd():
+                    return self.cache[(host, user, passwd)]
+            ftp = ftplib.FTP_TLS()
+            ftp.set_debuglevel(1 if logger.isEnabledFor(logging.DEBUG) else 0)
+            ftp.connect(host)
+            env_user = os.getenv("TOIL_FTP_USER")
+            env_passwd = os.getenv("TOIL_FTP_PASSWORD")
+            if env_user:
+                user = env_user
+            if env_passwd:
+                passwd = env_passwd
+            ftp.login(user or "", passwd or "", secure=not self.insecure)
+            self.cache[(host, user, passwd)] = ftp
+            return ftp
+        return None
+
+    def _recall_credentials(
+        self, desired_host: str
+    ) -> tuple[Optional[str], Optional[str]]:
+        for host, user, passwd in self.cache:
+            if desired_host == host:
+                return user, passwd
+        return None, None
+
+    def size(self, fn: str) -> Optional[int]:
+        ftp = self._connect(fn)
+        if ftp:
+            host, user, passwd, path = self._parse_url(fn)
+            try:
+                return ftp.size(path)
+            except ftplib.all_errors as e:
+                if str(e) == "550 SIZE not allowed in ASCII mode":
+                    # some servers don't allow grabbing size in ascii mode
+                    # https://stackoverflow.com/questions/22090001/get-folder-size-using-ftplib/22093848#22093848
+                    ftp.voidcmd("TYPE I")
+                    return ftp.size(path)
+                if host is None:
+                    # no host
+                    return None
+                handle = urlopen("ftp://{}:{}@{}/{}".format(user, passwd, host, path))
+                info = handle.info()
+                handle.close()
+                if "Content-length" in info:
+                    return int(info["Content-length"])
+                return None
+
+        return None

src/toil/lib/aws/utils.py

@@ -20,6 +20,7 @@
 from urllib.parse import ParseResult
 
 from toil.lib.aws import AWSRegionName, AWSServerErrors, session
+from toil.lib.conversions import strtobool
 from toil.lib.misc import printq
 from toil.lib.retry import (
     DEFAULT_DELAYS,
@@ -363,7 +364,7 @@ def get_object_for_url(url: ParseResult, existing: Optional[bool] = None) -> "S3
     host = os.environ.get("TOIL_S3_HOST", None)
     port = os.environ.get("TOIL_S3_PORT", None)
     protocol = "https"
-    if os.environ.get("TOIL_S3_USE_SSL", True) == "False":
+    if strtobool(os.environ.get("TOIL_S3_USE_SSL", 'True')) is False:
         protocol = "http"
     if host:
         endpoint_url = f"{protocol}://{host}" + f":{port}" if port else ""
@@ -425,7 +426,7 @@ def list_objects_for_url(url: ParseResult) -> list[str]:
     host = os.environ.get("TOIL_S3_HOST", None)
     port = os.environ.get("TOIL_S3_PORT", None)
     protocol = "https"
-    if os.environ.get("TOIL_S3_USE_SSL", True) == "False":
+    if strtobool(os.environ.get("TOIL_S3_USE_SSL", 'True')) is False:
         protocol = "http"
     if host:
         endpoint_url = f"{protocol}://{host}" + f":{port}" if port else ""