aws-cli/awscli/compat.py at develop · lovejavaee/aws-cli · GitHub

History

529 lines (434 loc) · 18.4 KB

Raw

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

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

# Licensed under the Apache License, Version 2.0 (the "License"). You

# may not use this file except in compliance with the License. A copy of

# the License is located at

# http://aws.amazon.com/apache2.0/

# or in the "license" file accompanying this file. This file is

# distributed on an "AS IS" BASIS, 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 sys

import re

import shlex

import os

import os.path

import platform

import zipfile

import signal

import contextlib

from configparser import RawConfigParser

from botocore.compat import six

#import botocore.compat

from botocore.compat import OrderedDict

# If you ever want to import from the vendored six. Add it here and then

# import from awscli.compat. Also try to keep it in alphabetical order.

# This may get large.

advance_iterator = six.advance_iterator

PY3 = six.PY3

queue = six.moves.queue

shlex_quote = six.moves.shlex_quote

StringIO = six.StringIO

BytesIO = six.BytesIO

urlopen = six.moves.urllib.request.urlopen

binary_type = six.binary_type

# Most, but not all, python installations will have zlib. This is required to

# compress any files we send via a push. If we can't compress, we can still

# package the files in a zip container.

try:

import zlib

ZIP_COMPRESSION_MODE = zipfile.ZIP_DEFLATED

except ImportError:

ZIP_COMPRESSION_MODE = zipfile.ZIP_STORED

try:

import sqlite3

except ImportError:

sqlite3 = None

is_windows = sys.platform == 'win32'

if is_windows:

default_pager = 'more'

else:

default_pager = 'less -R'

class StdinMissingError(Exception):

def __init__(self):

message = (

'stdin is required for this operation, but is not available.'

)

super(StdinMissingError, self).__init__(message)

class NonTranslatedStdout(object):

""" This context manager sets the line-end translation mode for stdout.

It is deliberately set to binary mode so that `\r` does not get added to

the line ending. This can be useful when printing commands where a

windows style line ending would cause errors.

"""

def __enter__(self):

if sys.platform == "win32":

import msvcrt

self.previous_mode = msvcrt.setmode(sys.stdout.fileno(),

os.O_BINARY)

return sys.stdout

def __exit__(self, type, value, traceback):

if sys.platform == "win32":

import msvcrt

msvcrt.setmode(sys.stdout.fileno(), self.previous_mode)

def ensure_text_type(s):

if isinstance(s, six.text_type):

return s

if isinstance(s, six.binary_type):

return s.decode('utf-8')

raise ValueError("Expected str, unicode or bytes, received %s." % type(s))

if six.PY3:

import collections.abc as collections_abc

import locale

import urllib.parse as urlparse

from urllib.error import URLError

raw_input = input

def get_binary_stdin():

if sys.stdin is None:

raise StdinMissingError()

return sys.stdin.buffer

def get_binary_stdout():

return sys.stdout.buffer

def _get_text_writer(stream, errors):

return stream

def compat_open(filename, mode='r', encoding=None):

"""Back-port open() that accepts an encoding argument.

In python3 this uses the built in open() and in python2 this

uses the io.open() function.

If the file is not being opened in binary mode, then we'll

use locale.getpreferredencoding() to find the preferred

encoding.

"""

if 'b' not in mode:

encoding = locale.getpreferredencoding()

return open(filename, mode, encoding=encoding)

def bytes_print(statement, stdout=None):

"""

This function is used to write raw bytes to stdout.

"""

if stdout is None:

stdout = sys.stdout

if getattr(stdout, 'buffer', None):

stdout.buffer.write(statement)

else:

# If it is not possible to write to the standard out buffer.

# The next best option is to decode and write to standard out.

stdout.write(statement.decode('utf-8'))

else:

import codecs

import collections as collections_abc

import locale

import io

import urlparse

from urllib2 import URLError

raw_input = raw_input

def get_binary_stdin():

if sys.stdin is None:

raise StdinMissingError()

return sys.stdin

def get_binary_stdout():

return sys.stdout

def _get_text_writer(stream, errors):

# In python3, all the sys.stdout/sys.stderr streams are in text

# mode. This means they expect unicode, and will encode the

# unicode automatically before actually writing to stdout/stderr.

# In python2, that's not the case. In order to provide a consistent

# interface, we can create a wrapper around sys.stdout that will take

# unicode, and automatically encode it to the preferred encoding.

# That way consumers can just call get_text_writer(stream) and write

# unicode to the returned stream. Note that get_text_writer

# just returns the stream in the PY3 section above because python3

# handles this.

# We're going to use the preferred encoding, but in cases that there is

# no preferred encoding we're going to fall back to assuming ASCII is

# what we should use. This will currently break the use of

# PYTHONIOENCODING, which would require checking stream.encoding first,

# however, the existing behavior is to only use

# locale.getpreferredencoding() and so in the hope of not breaking what

# is currently working, we will continue to only use that.

encoding = locale.getpreferredencoding()

if encoding is None:

encoding = "ascii"

return codecs.getwriter(encoding)(stream, errors)

def compat_open(filename, mode='r', encoding=None):

# See docstring for compat_open in the PY3 section above.

if 'b' not in mode:

encoding = locale.getpreferredencoding()

return io.open(filename, mode, encoding=encoding)

def bytes_print(statement, stdout=None):

if stdout is None:

stdout = sys.stdout

stdout.write(statement)

def get_stdout_text_writer():

return _get_text_writer(sys.stdout, errors="strict")

def get_stderr_text_writer():

return _get_text_writer(sys.stderr, errors="replace")

def compat_input(prompt):

"""

Cygwin's pty's are based on pipes. Therefore, when it interacts with a Win32

program (such as Win32 python), what that program sees is a pipe instead of

a console. This is important because python buffers pipes, and so on a

pty-based terminal, text will not necessarily appear immediately. In most

cases, this isn't a big deal. But when we're doing an interactive prompt,

the result is that the prompts won't display until we fill the buffer. Since

raw_input does not flush the prompt, we need to manually write and flush it.

See https://github.com/mintty/mintty/issues/56 for more details.

"""

sys.stdout.write(prompt)

sys.stdout.flush()

return raw_input()

def compat_shell_quote(s, platform=None):

"""Return a shell-escaped version of the string *s*

Unfortunately `shlex.quote` doesn't support Windows, so this method

provides that functionality.

"""

if platform is None:

platform = sys.platform

if platform == "win32":

return _windows_shell_quote(s)

else:

return shlex_quote(s)

def _windows_shell_quote(s):

"""Return a Windows shell-escaped version of the string *s*

Windows has potentially bizarre rules depending on where you look. When

spawning a process via the Windows C runtime the rules are as follows:

https://docs.microsoft.com/en-us/cpp/cpp/parsing-cpp-command-line-arguments

To summarize the relevant bits:

* Only space and tab are valid delimiters

* Double quotes are the only valid quotes

* Backslash is interpreted literally unless it is part of a chain that

leads up to a double quote. Then the backslashes escape the backslashes,

and if there is an odd number the final backslash escapes the quote.

:param s: A string to escape

:return: An escaped string

"""

if not s:

return '""'

buff = []

num_backspaces = 0

for character in s:

if character == '\\':

# We can't simply append backslashes because we don't know if

# they will need to be escaped. Instead we separately keep track

# of how many we've seen.

num_backspaces += 1

elif character == '"':

if num_backspaces > 0:

# The backslashes are part of a chain that lead up to a

# double quote, so they need to be escaped.

buff.append('\\' * (num_backspaces * 2))

num_backspaces = 0

# The double quote also needs to be escaped. The fact that we're

# seeing it at all means that it must have been escaped in the

# original source.

buff.append('\\"')

else:

if num_backspaces > 0:

# The backslashes aren't part of a chain leading up to a

# double quote, so they can be inserted directly without

# being escaped.

buff.append('\\' * num_backspaces)

num_backspaces = 0

buff.append(character)

# There may be some leftover backspaces if they were on the trailing

# end, so they're added back in here.

if num_backspaces > 0:

buff.append('\\' * num_backspaces)

new_s = ''.join(buff)

if ' ' in new_s or '\t' in new_s:

# If there are any spaces or tabs then the string needs to be double

# quoted.

return '"%s"' % new_s

return new_s

def get_popen_kwargs_for_pager_cmd(pager_cmd=None):

"""Returns the default pager to use dependent on platform

:rtype: str

:returns: A string represent the paging command to run based on the

platform being used.

"""

popen_kwargs = {}

if pager_cmd is None:

pager_cmd = default_pager

# Similar to what we do with the help command, we need to specify

# shell as True to make it work in the pager for Windows

if is_windows:

popen_kwargs = {'shell': True}

else:

pager_cmd = shlex.split(pager_cmd)

popen_kwargs['args'] = pager_cmd

return popen_kwargs

@contextlib.contextmanager

def ignore_user_entered_signals():

"""

Ignores user entered signals to avoid process getting killed.

"""

if is_windows:

signal_list = [signal.SIGINT]

else:

signal_list = [signal.SIGINT, signal.SIGQUIT, signal.SIGTSTP]

actual_signals = []

for user_signal in signal_list:

actual_signals.append(signal.signal(user_signal, signal.SIG_IGN))

try:

yield

finally:

for sig, user_signal in enumerate(signal_list):

signal.signal(user_signal, actual_signals[sig])

# linux_distribution is used by the CodeDeploy customization. Python 3.8

# removed it from the stdlib, so it is vendored here in the case where the

# import fails.

try:

from platform import linux_distribution

except ImportError:

_UNIXCONFDIR = '/etc'

def _dist_try_harder(distname, version, id):

""" Tries some special tricks to get the distribution

information in case the default method fails.

Currently supports older SuSE Linux, Caldera OpenLinux and

Slackware Linux distributions.

"""

if os.path.exists('/var/adm/inst-log/info'):

# SuSE Linux stores distribution information in that file

distname = 'SuSE'

with open('/var/adm/inst-log/info') as f:

for line in f:

tv = line.split()

if len(tv) == 2:

tag, value = tv

else:

continue

if tag == 'MIN_DIST_VERSION':

version = value.strip()

elif tag == 'DIST_IDENT':

values = value.split('-')

id = values[2]

return distname, version, id

if os.path.exists('/etc/.installed'):

# Caldera OpenLinux has some infos in that file (thanks to Colin Kong)

with open('/etc/.installed') as f:

for line in f:

pkg = line.split('-')

if len(pkg) >= 2 and pkg[0] == 'OpenLinux':

# XXX does Caldera support non Intel platforms ? If yes,

# where can we find the needed id ?

return 'OpenLinux', pkg[1], id

if os.path.isdir('/usr/lib/setup'):

# Check for slackware version tag file (thanks to Greg Andruk)

verfiles = os.listdir('/usr/lib/setup')

for n in range(len(verfiles)-1, -1, -1):

if verfiles[n][:14] != 'slack-version-':

del verfiles[n]

if verfiles:

verfiles.sort()

distname = 'slackware'

version = verfiles[-1][14:]

return distname, version, id

_release_filename = re.compile(r'(\w+)[-_](release|version)', re.ASCII)

_lsb_release_version = re.compile(r'(.+)'

r' release '

r'([\d.]+)'

r'[^(]*(?:\((.+)\))?', re.ASCII)

_release_version = re.compile(r'([^0-9]+)'

r'(?: release )?'

r'([\d.]+)'

r'[^(]*(?:\((.+)\))?', re.ASCII)

# See also http://www.novell.com/coolsolutions/feature/11251.html

# and http://linuxmafia.com/faq/Admin/release-files.html

# and http://data.linux-ntfs.org/rpm/whichrpm

# and http://www.die.net/doc/linux/man/man1/lsb_release.1.html

_supported_dists = (

'SuSE', 'debian', 'fedora', 'redhat', 'centos',

'mandrake', 'mandriva', 'rocks', 'slackware', 'yellowdog', 'gentoo',

'UnitedLinux', 'turbolinux', 'arch', 'mageia')

def _parse_release_file(firstline):

# Default to empty 'version' and 'id' strings. Both defaults are used

# when 'firstline' is empty. 'id' defaults to empty when an id can not

# be deduced.

version = ''

id = ''

# Parse the first line

m = _lsb_release_version.match(firstline)

if m is not None:

# LSB format: "distro release x.x (codename)"

return tuple(m.groups())

# Pre-LSB format: "distro x.x (codename)"

m = _release_version.match(firstline)

if m is not None:

return tuple(m.groups())

# Unknown format... take the first two words

l = firstline.strip().split()

if l:

version = l[0]

if len(l) > 1:

id = l[1]

return '', version, id

_distributor_id_file_re = re.compile("(?:DISTRIB_ID\s*=)\s*(.*)", re.I)

_release_file_re = re.compile("(?:DISTRIB_RELEASE\s*=)\s*(.*)", re.I)

_codename_file_re = re.compile("(?:DISTRIB_CODENAME\s*=)\s*(.*)", re.I)

def linux_distribution(distname='', version='', id='',

supported_dists=_supported_dists,

full_distribution_name=1):

return _linux_distribution(distname, version, id, supported_dists,

full_distribution_name)

def _linux_distribution(distname, version, id, supported_dists,

full_distribution_name):

""" Tries to determine the name of the Linux OS distribution name.

The function first looks for a distribution release file in

/etc and then reverts to _dist_try_harder() in case no

suitable files are found.

supported_dists may be given to define the set of Linux

distributions to look for. It defaults to a list of currently

supported Linux distributions identified by their release file

name.

If full_distribution_name is true (default), the full

distribution read from the OS is returned. Otherwise the short

name taken from supported_dists is used.

Returns a tuple (distname, version, id) which default to the

args given as parameters.

"""

# check for the Debian/Ubuntu /etc/lsb-release file first, needed so

# that the distribution doesn't get identified as Debian.

# https://bugs.python.org/issue9514

try:

with open("/etc/lsb-release", "r") as etclsbrel:

for line in etclsbrel:

m = _distributor_id_file_re.search(line)

if m:

_u_distname = m.group(1).strip()

m = _release_file_re.search(line)

if m:

_u_version = m.group(1).strip()

m = _codename_file_re.search(line)

if m:

_u_id = m.group(1).strip()

if _u_distname and _u_version:

return (_u_distname, _u_version, _u_id)

except (EnvironmentError, UnboundLocalError):

pass

try:

etc = os.listdir(_UNIXCONFDIR)

except OSError:

# Probably not a Unix system

return distname, version, id

etc.sort()

for file in etc:

m = _release_filename.match(file)

if m is not None:

_distname, dummy = m.groups()

if _distname in supported_dists:

distname = _distname

break

else:

return _dist_try_harder(distname, version, id)

# Read the first line

with open(os.path.join(_UNIXCONFDIR, file), 'r',

encoding='utf-8', errors='surrogateescape') as f:

firstline = f.readline()

_distname, _version, _id = _parse_release_file(firstline)

if _distname and full_distribution_name:

distname = _distname

if _version:

version = _version

if _id:

id = _id

return distname, version, id

aws-cli/awscli/compat.py at develop · lovejavaee/aws-cli · GitHub

Expand file tree

Breadcrumbs

compat.py

Latest commit

History

Breadcrumbs

compat.py

File metadata and controls