Source code for linkcheck.checker.ftpurl

# Copyright (C) 2000-2014 Bastian Kleineidam
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
Handle FTP links.

import ftplib
from io import StringIO

from .. import log, LOG_CHECK, LinkCheckerError, mimeutil
from . import internpaturl, get_index_html
from .const import WARN_FTP_MISSING_SLASH

[docs]class FtpUrl(internpaturl.InternPatternUrl): """ Url link with ftp scheme. """
[docs] def reset(self): """ Initialize FTP url data. """ super().reset() # list of files for recursion self.files = [] # last part of URL filename self.filename = None self.filename_encoding = 'iso-8859-1'
[docs] def check_connection(self): """ Check in this order: login, changing directory, list the file. """ self.login() self.negotiate_encoding() self.filename = self.cwd() self.listfile() self.files = [] return None
[docs] def login(self): """Log into ftp server and check the welcome message.""" self.url_connection = ftplib.FTP(timeout=self.aggregate.config["timeout"]) if log.is_debug(LOG_CHECK): self.url_connection.set_debuglevel(1) try: self.url_connection.connect(, self.port) _user, _password = self.get_user_password() if _user is None: self.url_connection.login() elif _password is None: self.url_connection.login(_user) else: self.url_connection.login(_user, _password) info = self.url_connection.getwelcome() if info: # note that the info may change every time a user logs in, # so don't add it to the url_data info. log.debug(LOG_CHECK, "FTP info %s", info) else: raise LinkCheckerError(_("Got no answer from FTP server")) except EOFError as msg: raise LinkCheckerError( _("Remote host has closed connection: %(msg)s") % str(msg) )
[docs] def negotiate_encoding(self): """Check if server can handle UTF-8 encoded filenames. See also RFC 2640.""" try: features = self.url_connection.sendcmd("FEAT") except ftplib.error_perm as msg: log.debug(LOG_CHECK, "Ignoring error when getting FTP features: %s" % msg) else: log.debug(LOG_CHECK, "FTP features %s", features) if " UTF-8" in features.splitlines(): self.filename_encoding = "utf-8"
[docs] def cwd(self): """ Change to URL parent directory. Return filename of last path component. """ path = self.urlparts[2] if isinstance(path, bytes): path = path.decode(self.filename_encoding, 'replace') dirname = path.strip('/') dirs = dirname.split('/') filename = dirs.pop() self.url_connection.cwd('/') for d in dirs: self.url_connection.cwd(d) return filename
[docs] def listfile(self): """ See if filename is in the current FTP directory. """ if not self.filename: return files = self.get_files() log.debug(LOG_CHECK, "FTP files %s", str(files)) if self.filename in files: # file found return # it could be a directory if the trailing slash was forgotten if f"{self.filename}/" in files: if not self.url.endswith('/'): self.add_warning( _("Missing trailing directory slash in ftp url."), tag=WARN_FTP_MISSING_SLASH, ) self.url += '/' return raise ftplib.error_perm("550 File not found")
[docs] def get_files(self): """Get list of filenames in directory. Subdirectories have an ending slash.""" files = [] def add_entry(line): """Parse list line and add the entry it points to to the file list.""" log.debug(LOG_CHECK, "Directory entry %r", line) from ..ftpparse import ftpparse fpo = ftpparse(line) if fpo is not None and fpo["name"]: name = fpo["name"] if fpo["trycwd"]: name += "/" if fpo["trycwd"] or fpo["tryretr"]: files.append(name) self.url_connection.dir(add_entry) return files
[docs] def is_parseable(self): """See if URL target is parseable for recursion.""" if self.is_directory(): return True return self.is_content_type_parseable()
[docs] def is_directory(self): """See if URL target is a directory.""" # either the path is empty, or ends with a slash path = self.urlparts[2] return (not path) or path.endswith('/')
[docs] def set_content_type(self): """Set URL content type, or an empty string if content type could not be found.""" self.content_type = mimeutil.guess_mimetype(self.url, read=self.get_content) log.debug(LOG_CHECK, "MIME type: %s", self.content_type)
[docs] def read_content(self): """Return URL target content, or in case of directories a dummy HTML file with links to the files.""" if self.is_directory(): self.url_connection.cwd(self.filename) self.files = self.get_files() # XXX limit number of files? data = get_index_html(self.files) else: # download file in BINARY mode ftpcmd = f"RETR {self.filename}" buf = StringIO() def stor_data(s): """Helper method storing given data""" # limit the download size if (buf.tell() + len(s)) > self.max_size: raise LinkCheckerError(_("FTP file size too large")) buf.write(s) self.url_connection.retrbinary(ftpcmd, stor_data) data = buf.getvalue() buf.close() return data
[docs] def close_connection(self): """Release the open connection from the connection pool.""" if self.url_connection is not None: try: self.url_connection.quit() except Exception: pass self.url_connection = None