1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
#!/usr/bin/python3
# -*- mode: python -*-
#
# This file is part of FreedomBox.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 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
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
"""
Actions for sshfs.
"""
import argparse
import json
import os
import subprocess
import sys
TIMEOUT = 30
class AlreadyMountedError(Exception):
"""Exception raised when mount point is already mounted."""
def parse_arguments():
"""Return parsed command line arguments as dictionary."""
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')
mount = subparsers.add_parser('mount', help='mount an ssh filesystem')
mount.add_argument('--mountpoint', help='Local mountpoint', required=True)
mount.add_argument('--path', help='Remote ssh path to mount',
required=True)
mount.add_argument('--ssh-keyfile', help='Path of private ssh key',
default=None, required=False)
umount = subparsers.add_parser('umount', help='unmount an ssh filesystem')
umount.add_argument('--mountpoint', help='Mountpoint to unmount',
required=True)
is_mounted = subparsers.add_parser(
'is-mounted', help='Check whether a mountpoint is mounted')
is_mounted.add_argument('--mountpoint', help='Mountpoint to check',
required=True)
subparsers.required = True
return parser.parse_args()
def subcommand_mount(arguments):
"""Mount a remote ssh path via sshfs."""
try:
validate_mountpoint(arguments.mountpoint)
except AlreadyMountedError:
return
remote_path = arguments.path
kwargs = {}
# the shell would expand ~/ to the local home directory
remote_path = remote_path.replace('~/', '').replace('~', '')
cmd = [
'sshfs', remote_path, arguments.mountpoint, '-o',
'UserKnownHostsFile=/dev/null', '-o', 'StrictHostKeyChecking=no'
]
if arguments.ssh_keyfile:
cmd += ['-o', 'IdentityFile=' + arguments.ssh_keyfile]
else:
password = read_password()
if not password:
raise ValueError('mount requires either a password or ssh_keyfile')
cmd += ['-o', 'password_stdin']
kwargs['input'] = password.encode()
subprocess.run(cmd, check=True, timeout=TIMEOUT, **kwargs)
def subcommand_umount(arguments):
"""Unmount a mountpoint."""
subprocess.run(['umount', arguments.mountpoint], check=True)
def validate_mountpoint(mountpoint):
"""Check that the folder is empty, and create it if it doesn't exist"""
if os.path.exists(mountpoint):
if _is_mounted(mountpoint):
raise AlreadyMountedError(
'Mountpoint %s already mounted' % mountpoint)
if os.listdir(mountpoint) or not os.path.isdir(mountpoint):
raise ValueError(
'Mountpoint %s is not an empty directory' % mountpoint)
else:
os.makedirs(mountpoint)
def _is_mounted(mountpoint):
"""Return boolean whether a local directory is a mountpoint."""
cmd = ['mountpoint', '-q', mountpoint]
# mountpoint exits with status non-zero if it didn't find a mountpoint
try:
subprocess.run(cmd, check=True)
return True
except subprocess.CalledProcessError:
return False
def subcommand_is_mounted(arguments):
"""Print whether a path is already mounted."""
print(json.dumps(_is_mounted(arguments.mountpoint)))
def read_password():
"""Read the password from stdin."""
if sys.stdin.isatty():
return ''
return ''.join(sys.stdin)
def main():
"""Parse arguments and perform all duties."""
arguments = parse_arguments()
subcommand = arguments.subcommand.replace('-', '_')
subcommand_method = globals()['subcommand_' + subcommand]
subcommand_method(arguments)
if __name__ == '__main__':
main()
|
