9
9
import requests
10
10
import unix_ar
11
11
12
- from aws_gate .constants import DEFAULT_GATE_BIN_PATH , PLUGIN_INSTALL_PATH , PLUGIN_NAME , SSM_PLUGIN_PATH
12
+ from aws_gate .constants import (
13
+ DEFAULT_GATE_BIN_PATH ,
14
+ PLUGIN_INSTALL_PATH ,
15
+ PLUGIN_NAME ,
16
+ SSM_PLUGIN_PATH ,
17
+ )
13
18
from aws_gate .exceptions import UnsupportedPlatormError
14
19
from aws_gate .utils import execute
15
20
16
21
logger = logging .getLogger (__name__ )
17
22
18
23
19
24
def _check_plugin_version (path = PLUGIN_INSTALL_PATH ):
20
- return execute (path , [' --version' ], capture_output = True )
25
+ return execute (path , [" --version" ], capture_output = True )
21
26
22
27
23
28
class Plugin :
@@ -26,7 +31,7 @@ class Plugin:
26
31
27
32
@property
28
33
def is_installed (self ):
29
- logger .debug (' Checking if %s exists and is executable' , PLUGIN_INSTALL_PATH )
34
+ logger .debug (" Checking if %s exists and is executable" , PLUGIN_INSTALL_PATH )
30
35
return shutil .which (PLUGIN_INSTALL_PATH ) is None
31
36
32
37
def download (self ):
@@ -35,71 +40,81 @@ def download(self):
35
40
36
41
self .download_path = os .path .join (tmp_dir , file_name )
37
42
try :
38
- logger .debug (' Downloading session-manager-plugin archive from %s' , self .url )
43
+ logger .debug (" Downloading session-manager-plugin archive from %s" , self .url )
39
44
with requests .get (self .url , stream = True ) as req :
40
45
req .raise_for_status ()
41
- with open (self .download_path , 'wb' ) as f :
46
+ with open (self .download_path , "wb" ) as f :
42
47
shutil .copyfileobj (req .raw , f )
43
- logger .debug (' Download stored at %s' , self .download_path )
48
+ logger .debug (" Download stored at %s" , self .download_path )
44
49
except requests .exceptions .HTTPError as e :
45
- logger .error (' HTTP error while downloading %s: %s' , self .url , e )
50
+ logger .error (" HTTP error while downloading %s: %s" , self .url , e )
46
51
47
52
def install (self ):
48
53
download_dir = os .path .split (self .download_path )[0 ]
49
- plugin_src_path = os .path .join (download_dir , SSM_PLUGIN_PATH [platform .system ()]['bundle' ])
54
+ plugin_src_path = os .path .join (
55
+ download_dir , SSM_PLUGIN_PATH [platform .system ()]["bundle" ]
56
+ )
50
57
plugin_dst_path = PLUGIN_INSTALL_PATH
51
58
52
59
if not os .path .exists (DEFAULT_GATE_BIN_PATH ):
53
- logger .debug (' Creating %s' , DEFAULT_GATE_BIN_PATH )
60
+ logger .debug (" Creating %s" , DEFAULT_GATE_BIN_PATH )
54
61
os .makedirs (DEFAULT_GATE_BIN_PATH )
55
62
56
- with open (plugin_src_path , 'rb' ) as f_src , open (plugin_dst_path , 'wb' ) as f_dst :
57
- logger .debug (' Copying %s to %s' , plugin_src_path , plugin_dst_path )
63
+ with open (plugin_src_path , "rb" ) as f_src , open (plugin_dst_path , "wb" ) as f_dst :
64
+ logger .debug (" Copying %s to %s" , plugin_src_path , plugin_dst_path )
58
65
shutil .copyfileobj (f_src , f_dst )
59
66
60
- logger .debug (' Setting execution permissions on %s' , plugin_dst_path )
67
+ logger .debug (" Setting execution permissions on %s" , plugin_dst_path )
61
68
os .chmod (plugin_dst_path , 0o755 )
62
69
63
70
version = _check_plugin_version (PLUGIN_INSTALL_PATH )
64
- print (' {} (version {}) installed successfully!' .format (PLUGIN_NAME , version ))
71
+ print (" {} (version {}) installed successfully!" .format (PLUGIN_NAME , version ))
65
72
66
73
def extract (self ):
67
74
raise NotImplementedError
68
75
69
76
70
77
class MacPlugin (Plugin ):
71
- url = SSM_PLUGIN_PATH [' Darwin' ][ ' download' ]
78
+ url = SSM_PLUGIN_PATH [" Darwin" ][ " download" ]
72
79
73
80
def extract (self ):
74
81
if not zipfile .is_zipfile (self .download_path ):
75
- raise ValueError ('Invalid macOS session-manager-plugin ZIP file found {}' .format (self .download_path ))
82
+ raise ValueError (
83
+ "Invalid macOS session-manager-plugin ZIP file found {}" .format (
84
+ self .download_path
85
+ )
86
+ )
76
87
77
- with zipfile .ZipFile (self .download_path , 'r' ) as zip_file :
88
+ with zipfile .ZipFile (self .download_path , "r" ) as zip_file :
78
89
download_dir = os .path .split (self .download_path )[0 ]
79
- logger .debug ('Extracting session-manager-plugin archive at %s' , download_dir )
90
+ logger .debug (
91
+ "Extracting session-manager-plugin archive at %s" , download_dir
92
+ )
80
93
zip_file .extractall (download_dir )
81
94
82
95
83
96
class LinuxPlugin (Plugin ):
84
- url = SSM_PLUGIN_PATH [' Linux' ][ ' download' ]
97
+ url = SSM_PLUGIN_PATH [" Linux" ][ " download" ]
85
98
86
99
def extract (self ):
87
100
ar_file = unix_ar .open (self .download_path )
88
- tarball = ar_file .open (' data.tar.gz/' )
101
+ tarball = ar_file .open (" data.tar.gz/" )
89
102
tar_file = tarfile .open (fileobj = tarball )
90
103
download_dir = os .path .split (self .download_path )[0 ]
91
- logger .debug (' Extracting session-manager-plugin archive at %s' , download_dir )
104
+ logger .debug (" Extracting session-manager-plugin archive at %s" , download_dir )
92
105
tar_file .extractall (download_dir )
93
106
94
107
95
108
def bootstrap (force = False ):
96
109
system = platform .system ()
97
- if system == ' Darwin' :
110
+ if system == " Darwin" :
98
111
plugin = MacPlugin ()
99
- elif system == ' Linux' :
112
+ elif system == " Linux" :
100
113
plugin = LinuxPlugin ()
101
114
else :
102
- raise UnsupportedPlatormError ('Unable to bootstrap session-manager-plugin on {}' .format (system ))
115
+ raise UnsupportedPlatormError (
116
+ "Unable to bootstrap session-manager-plugin on {}" .format (system )
117
+ )
103
118
104
119
if plugin .is_installed or force :
105
120
plugin .download ()
0 commit comments