-
-
Notifications
You must be signed in to change notification settings - Fork 142
/
main.py
executable file
·3795 lines (3603 loc) · 149 KB
/
main.py
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
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
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# This file is part of Xpra.
# Copyright (C) 2011 Serviware (Arthur Huillet, <ahuillet@serviware.com>)
# Copyright (C) 2010-2021 Antoine Martin <antoine@xpra.org>
# Copyright (C) 2008 Nathaniel Smith <njs@pobox.com>
# Xpra is released under the terms of the GNU GPL v2, or, at your option, any
# later version. See the file COPYING for details.
import sys
import os.path
import stat
import socket
import time
import logging
from subprocess import Popen, PIPE, TimeoutExpired
import signal
import shlex
import traceback
from xpra import __version__ as XPRA_VERSION
from xpra.platform.dotxpra import DotXpra
from xpra.util import (
csv, envbool, envint, nonl, pver,
parse_simple_dict, noerr, sorted_nicely, typedict,
DEFAULT_PORT, DEFAULT_PORTS,
)
from xpra.exit_codes import (
EXIT_STR,
EXIT_OK, EXIT_FAILURE, EXIT_UNSUPPORTED, EXIT_CONNECTION_FAILED,
EXIT_NO_DISPLAY,
EXIT_CONNECTION_LOST, EXIT_REMOTE_ERROR,
EXIT_INTERNAL_ERROR, EXIT_FILE_TOO_BIG,
)
from xpra.os_util import (
get_util_logger, getuid, getgid, get_username_for_uid,
monotonic_time, bytestostr, use_tty,
WIN32, OSX, POSIX, SIGNAMES, is_Ubuntu,
)
from xpra.scripts.parsing import (
info, warn, error,
parse_vsock, parse_env, is_local,
fixup_defaults, validated_encodings, validate_encryption, do_parse_cmdline, show_sound_codec_help,
supports_shadow, supports_server, supports_proxy, supports_mdns,
)
from xpra.scripts.config import (
OPTION_TYPES, TRUE_OPTIONS, FALSE_OPTIONS, OFF_OPTIONS,
CLIENT_OPTIONS, NON_COMMAND_LINE_OPTIONS, CLIENT_ONLY_OPTIONS,
START_COMMAND_OPTIONS, BIND_OPTIONS, PROXY_START_OVERRIDABLE_OPTIONS, OPTIONS_ADDED_SINCE_V3, OPTIONS_COMPAT_NAMES,
InitException, InitInfo, InitExit,
fixup_options, dict_to_validated_config, get_xpra_defaults_dirs, get_defaults, read_xpra_conf,
make_defaults_struct, parse_bool, has_sound_support, name_to_field,
)
from xpra.log import is_debug_enabled, Logger, get_debug_args
assert info and warn and error, "used by modules importing those from here"
NO_ROOT_WARNING = envbool("XPRA_NO_ROOT_WARNING", False)
CLIPBOARD_CLASS = os.environ.get("XPRA_CLIPBOARD_CLASS")
WAIT_SERVER_TIMEOUT = envint("WAIT_SERVER_TIMEOUT", 90)
CONNECT_TIMEOUT = envint("XPRA_CONNECT_TIMEOUT", 20)
OPENGL_PROBE_TIMEOUT = envint("XPRA_OPENGL_PROBE_TIMEOUT", 5)
SYSTEMD_RUN = envbool("XPRA_SYSTEMD_RUN", True)
VERIFY_X11_SOCKET_TIMEOUT = envint("XPRA_VERIFY_X11_SOCKET_TIMEOUT", 1)
LIST_REPROBE_TIMEOUT = envint("XPRA_LIST_REPROBE_TIMEOUT", 10)
def nox():
DISPLAY = os.environ.get("DISPLAY")
if DISPLAY is not None:
del os.environ["DISPLAY"]
# This is an error on Fedora/RH, so make it an error everywhere so it will
# be noticed:
import warnings
warnings.filterwarnings("error", "could not open display")
return DISPLAY
def werr(*msg):
for x in msg:
noerr(sys.stderr.write, "%s\n" % (x,))
noerr(sys.stderr.flush)
def main(script_file, cmdline):
ml = envint("XPRA_MEM_USAGE_LOGGER")
if ml>0:
from xpra.util import start_mem_watcher
start_mem_watcher(ml)
if sys.flags.optimize>0: # pragma: no cover
sys.stderr.write("************************************************************\n")
sys.stderr.write("Warning: the python optimize flag is set to %i\n" % sys.flags.optimize)
sys.stderr.write(" xpra is very likely to crash\n")
sys.stderr.write("************************************************************\n")
time.sleep(5)
from xpra.platform import clean as platform_clean, command_error, command_info
if len(cmdline)==1:
cmdline.append("gui")
#turn off gdk scaling to make sure we get the actual window geometry:
os.environ["GDK_SCALE"]="1"
os.environ["GDK_DPI_SCALE"] = "1"
#client side decorations break window geometry,
#disable this "feature" unless explicitly enabled:
if os.environ.get("GTK_CSD") is None:
os.environ["GTK_CSD"] = "0"
if POSIX and not OSX and os.environ.get("XDG_SESSION_TYPE", "x11")=="x11" and not os.environ.get("GDK_BACKEND"):
os.environ["GDK_BACKEND"] = "x11"
if envbool("XPRA_NOMD5", False):
import hashlib
def nomd5(*_args):
raise ValueError("md5 support is disabled")
hashlib.algorithms_available.remove("md5")
hashlib.md5 = nomd5
def debug_exc(msg="run_mode error"):
get_util_logger().debug(msg, exc_info=True)
try:
defaults = make_defaults_struct()
fixup_defaults(defaults)
options, args = do_parse_cmdline(cmdline, defaults)
if not args:
raise InitExit(-1, "xpra: need a mode")
mode = args.pop(0)
def err(*args):
raise InitException(*args)
return run_mode(script_file, err, options, args, mode, defaults)
except SystemExit as e:
debug_exc()
return e.code
except InitExit as e:
debug_exc()
if str(e) and e.args and (e.args[0] or len(e.args)>1):
command_info("%s" % e)
return e.status
except InitInfo as e:
debug_exc()
command_info("%s" % e)
return 0
except InitException as e:
debug_exc()
command_error("xpra initialization error:\n %s" % e)
return 1
except AssertionError as e:
debug_exc()
command_error("xpra initialization error:\n %s" % e)
traceback.print_tb(sys.exc_info()[2])
return 1
except Exception:
debug_exc()
command_error("xpra main error:\n%s" % traceback.format_exc())
return 1
finally:
platform_clean()
def closestd(std):
if std:
try:
std.close()
except OSError: # pragma: no cover
pass
closestd(sys.stdout)
closestd(sys.stderr)
def configure_logging(options, mode):
if mode in (
"showconfig", "info", "id", "attach", "listen", "launcher", "stop", "print",
"control", "list", "list-windows", "list-mdns", "sessions", "mdns-gui", "bug-report",
"displays", "list-sessions", "wminfo", "wmname", "recover",
"clean-displays", "clean-sockets",
"splash", "qrcode",
"opengl-test",
"test-connect",
"encoding", "webcam", "clipboard-test",
"keyboard", "keyboard-test", "keymap", "gui-info", "network-info", "path-info",
"printing-info", "version-info", "gtk-info",
"colors-test", "colors-gradient-test", "transparent-colors", "transparent-window",
"auth",
):
s = sys.stdout
else:
s = sys.stderr
to = s
try:
import codecs
#print("locale.getpreferredencoding()=%s" % (locale.getpreferredencoding(),))
#python3 has a buffer attribute,
#which we must use if we want to be able to write bytes:
sbuf = getattr(s, "buffer", s)
to = codecs.getwriter("utf-8")(sbuf, "replace")
except Exception: # pragma: no cover
pass
#a bit naughty here, but it's easier to let xpra.log initialize
#the logging system every time, and just undo things here..
from xpra.log import (
setloghandler, enable_color, enable_format,
LOG_FORMAT, NOPREFIX_FORMAT,
SIGPIPEStreamHandler,
)
setloghandler(SIGPIPEStreamHandler(to))
if mode in (
"start", "start-desktop", "upgrade", "upgrade-desktop",
"attach", "listen", "shadow", "proxy",
"_sound_record", "_sound_play",
"stop", "print", "showconfig",
"request-start", "request-start-desktop", "request-shadow",
"_dialog", "_pass",
"pinentry",
):
if "help" in options.speaker_codec or "help" in options.microphone_codec:
server_mode = mode not in ("attach", "listen")
codec_help = show_sound_codec_help(server_mode, options.speaker_codec, options.microphone_codec)
raise InitInfo("\n".join(codec_help))
fmt = LOG_FORMAT
if mode in ("stop", "showconfig"):
fmt = NOPREFIX_FORMAT
if envbool("XPRA_COLOR_LOG", hasattr(to, "fileno") and os.isatty(to.fileno())):
enable_color(to, fmt)
else:
enable_format(fmt)
from xpra.log import add_debug_category, add_disabled_category, enable_debug_for, disable_debug_for
if options.debug:
categories = options.debug.split(",")
for cat in categories:
if not cat:
continue
if cat[0]=="-":
add_disabled_category(cat[1:])
disable_debug_for(cat[1:])
else:
add_debug_category(cat)
enable_debug_for(cat)
#always log debug level, we just use it selectively (see above)
logging.root.setLevel(logging.INFO)
def configure_network(options):
from xpra.net import compression, packet_encoding
ecs = compression.get_enabled_compressors()
for c in compression.ALL_COMPRESSORS:
enabled = c in ecs and c in options.compressors
setattr(compression, "use_%s" % c, enabled)
if not ecs:
#force compression level to zero since we have no compressors available:
options.compression_level = 0
ees = packet_encoding.get_enabled_encoders()
count = 0
for pe in packet_encoding.ALL_ENCODERS:
enabled = pe in ees and pe in options.packet_encoders
setattr(packet_encoding, "use_%s" % pe, enabled)
count += int(enabled)
#verify that at least one encoder is available:
if not count:
raise InitException("at least one valid packet encoder must be enabled")
def configure_env(env_str):
if env_str:
env = parse_env(env_str)
if POSIX and getuid()==0:
#running as root!
#sanitize: only allow "safe" environment variables
#as these may have been specified by a non-root user
env = dict((k,v) for k,v in env.items() if k.startswith("XPRA_"))
os.environ.update(env)
def systemd_run_command(mode, systemd_run_args=None, user=True):
cmd = ["systemd-run", "--description" , "xpra-%s" % mode, "--scope"]
if user:
cmd.append("--user")
LOG_SYSTEMD_WRAP = envbool("XPRA_LOG_SYSTEMD_WRAP", True)
if not LOG_SYSTEMD_WRAP:
cmd.append("--quiet")
if systemd_run_args:
cmd += shlex.split(systemd_run_args)
return cmd
def systemd_run_wrap(mode, args, systemd_run_args=None, **kwargs):
cmd = systemd_run_command(mode, systemd_run_args)
cmd += args
cmd.append("--systemd-run=no")
stderr = sys.stderr
LOG_SYSTEMD_WRAP = envbool("XPRA_LOG_SYSTEMD_WRAP", True)
if LOG_SYSTEMD_WRAP:
noerr(stderr.write, "using systemd-run to wrap '%s' server command\n" % mode)
LOG_SYSTEMD_WRAP_COMMAND = envbool("XPRA_LOG_SYSTEMD_WRAP_COMMAND", False)
if LOG_SYSTEMD_WRAP_COMMAND:
noerr(stderr.write, "%s\n" % " ".join(["'%s'" % x for x in cmd]))
try:
p = Popen(cmd, **kwargs)
return p.wait()
except KeyboardInterrupt:
return 128+signal.SIGINT
def isdisplaytype(args, *dtypes) -> bool:
if not args:
return False
d = args[0]
return any((d.startswith("%s/" % dtype) or d.startswith("%s:" % dtype) for dtype in dtypes))
def check_gtk():
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
assert Gtk
r = Gtk.init_check(None)
if not r[0]:
raise InitExit(EXIT_NO_DISPLAY, "failed to initialize Gtk, no display?")
check_display()
def check_display():
from xpra.platform.gui import can_access_display
if not can_access_display(): # pragma: no cover
raise InitExit(EXIT_NO_DISPLAY, "cannot access display")
def use_systemd_run(s):
if not SYSTEMD_RUN or not POSIX:
return False # pragma: no cover
systemd_run = parse_bool("systemd-run", s)
if systemd_run in (True, False):
return systemd_run
#detect if we should use it:
if is_Ubuntu() and (os.environ.get("SSH_TTY") or os.environ.get("SSH_CLIENT")): # pragma: no cover
#would fail
return False
from xpra.os_util import is_systemd_pid1
if not is_systemd_pid1():
return False # pragma: no cover
#test it:
cmd = ["systemd-run", "--quiet", "--user", "--scope", "--", "true"]
proc = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=False)
try:
proc.communicate(timeout=2)
r = proc.returncode
except TimeoutExpired: # pragma: no cover
r = None
if r is None:
try:
proc.terminate()
except Exception:
pass
try:
proc.communicate(timeout=1)
except TimeoutExpired: # pragma: no cover
r = None
return r==0
def run_mode(script_file, error_cb, options, args, mode, defaults):
#configure default logging handler:
if POSIX and getuid()==0 and options.uid==0 and mode!="proxy" and not NO_ROOT_WARNING:
warn("\nWarning: running as root")
display_is_remote = isdisplaytype(args, "ssh", "tcp", "ssl", "vsock")
if mode in ("start", "start-desktop", "upgrade", "upgrade-desktop", "shadow") and not display_is_remote:
if use_systemd_run(options.systemd_run):
#make sure we run via the same interpreter,
#inject it into the command line if we have to:
argv = list(sys.argv)
if argv[0].find("python")<0:
argv.insert(0, "python%i.%i" % (sys.version_info.major, sys.version_info.minor))
return systemd_run_wrap(mode, argv, options.systemd_run_args)
configure_env(options.env)
configure_logging(options, mode)
configure_network(options)
if mode not in ("showconfig", "splash") and POSIX and not OSX and os.environ.get("XDG_RUNTIME_DIR") is None and getuid()>0:
xrd = "/run/user/%i" % getuid()
if os.path.exists(xrd):
warn("Warning: using '%s' as XDG_RUNTIME_DIR" % xrd)
os.environ["XDG_RUNTIME_DIR"] = xrd
else:
warn("Warning: XDG_RUNTIME_DIR is not defined")
warn(" and '%s' does not exist" % xrd)
if os.path.exists("/tmp") and os.path.isdir("/tmp"):
xrd = "/tmp"
warn(" using '%s'" % xrd)
os.environ["XDG_RUNTIME_DIR"] = xrd
if not mode.startswith("_sound_"):
#only the sound subcommands should ever actually import GStreamer:
try:
from xpra.sound.gstreamer_util import prevent_import
prevent_import()
except ImportError: # pragma: no cover
pass
#sound commands don't want to set the name
#(they do it later to prevent glib import conflicts)
#"attach" does it when it received the session name from the server
if mode not in ("attach", "listen", "start", "start-desktop", "upgrade", "upgrade-desktop", "proxy", "shadow"):
from xpra.platform import set_name
set_name("Xpra", "Xpra %s" % mode.strip("_"))
if mode in (
"start", "start-desktop",
"shadow", "attach", "listen",
"upgrade", "upgrade-desktop",
"recover",
"request-start", "request-start-desktop", "request-shadow",
):
options.encodings = validated_encodings(options.encodings)
try:
return do_run_mode(script_file, error_cb, options, args, mode, defaults)
except KeyboardInterrupt as e:
info("\ncaught %s, exiting" % repr(e))
return 128+signal.SIGINT
def do_run_mode(script_file, error_cb, options, args, mode, defaults):
display_is_remote = isdisplaytype(args, "ssh", "tcp", "ssl", "vsock")
if mode in ("start", "start-desktop", "shadow") and display_is_remote:
#ie: "xpra start ssh://USER@HOST:SSHPORT/DISPLAY --start-child=xterm"
return run_remote_server(error_cb, options, args, mode, defaults)
if mode in ("start", "start-desktop") and args and parse_bool("attach", options.attach) is True:
assert not display_is_remote
#maybe the server is already running
#and we don't need to bother trying to start it:
try:
display = pick_display(error_cb, options, args)
except Exception:
pass
else:
dotxpra = DotXpra(options.socket_dir, options.socket_dirs)
display_name = display.get("display_name")
if display_name:
state = dotxpra.get_display_state(display_name)
if state==DotXpra.LIVE:
noerr(sys.stdout.write, "existing live display found, attaching")
return do_run_mode(script_file, error_cb, options, args, "attach", defaults)
if mode in ("start", "start-desktop", "upgrade", "upgrade-desktop", "shadow", "proxy"):
return run_server(script_file, error_cb, options, args, mode, defaults)
elif mode in (
"attach", "listen", "detach",
"screenshot", "version", "info", "id",
"control", "_monitor", "shell", "print",
"qrcode",
"connect-test", "request-start", "request-start-desktop", "request-shadow",
):
return run_client(error_cb, options, args, mode)
elif mode in ("stop", "exit"):
return run_stopexit(mode, error_cb, options, args)
elif mode == "top":
return run_top(error_cb, options, args)
elif mode == "list":
return run_list(error_cb, options, args)
elif mode == "list-windows":
return run_list_windows(error_cb, options, args)
elif mode == "list-mdns" and supports_mdns:
return run_list_mdns(error_cb, args)
elif mode == "mdns-gui" and supports_mdns:
check_gtk()
return run_mdns_gui(error_cb, options)
elif mode == "list-sessions":
return run_list_sessions(args, options)
elif mode == "sessions":
check_gtk()
return run_sessions_gui(error_cb, options)
elif mode == "displays":
check_gtk()
return run_displays(args)
elif mode == "clean-displays":
return run_clean_displays(args)
elif mode == "clean-sockets":
return run_clean_sockets(options, args)
elif mode=="recover":
return run_recover(script_file, error_cb, options, args, defaults)
elif mode == "wminfo":
return run_wminfo(args)
elif mode == "wmname":
return run_wmname(args)
elif mode == "launcher":
check_gtk()
from xpra.client.gtk_base.client_launcher import main as launcher_main
return launcher_main(["xpra"]+args)
elif mode == "gui":
check_gtk()
from xpra.gtk_common import gui
return gui.main()
elif mode == "start-gui":
check_gtk()
from xpra.gtk_common import start_gui
return start_gui.main(options)
elif mode == "bug-report":
check_gtk()
from xpra.scripts import bug_report
bug_report.main(["xpra"]+args)
elif (
mode=="_proxy" or
(mode in ("_proxy_start", "_proxy_start_desktop") and supports_server) or
(mode=="_proxy_shadow_start" and supports_shadow)
):
nox()
return run_proxy(error_cb, options, script_file, args, mode, defaults)
elif mode in ("_sound_record", "_sound_play", "_sound_query"):
if not has_sound_support():
error_cb("no sound support!")
from xpra.sound.wrapper import run_sound
return run_sound(mode, error_cb, options, args)
elif mode=="pinentry":
check_gtk()
return run_pinentry(args)
elif mode=="_dialog":
check_gtk()
return run_dialog(args)
elif mode=="_pass":
check_gtk()
return run_pass(args)
elif mode=="send-file":
check_gtk()
return run_send_file(args)
elif mode=="splash":
check_gtk()
return run_splash(args)
elif mode=="opengl":
check_gtk()
return run_glcheck(options)
elif mode=="opengl-probe":
check_gtk()
return run_glprobe(options)
elif mode=="opengl-test":
check_gtk()
return run_glprobe(options, True)
elif mode=="encoding":
from xpra.codecs import loader
return loader.main(args)
elif mode=="video":
from xpra.codecs import video_helper
return video_helper.main()
elif mode=="nvinfo":
from xpra.codecs import nv_util
return nv_util.main()
elif mode=="webcam":
check_gtk()
from xpra.scripts import show_webcam
return show_webcam.main()
elif mode=="webcam-info":
from xpra.platform import webcam
webcam.main(args)
elif mode=="clipboard-test":
check_gtk()
from xpra.gtk_common import gtk_view_clipboard
return gtk_view_clipboard.main()
elif mode=="keyboard":
from xpra.platform import keyboard
return keyboard.main()
elif mode=="keyboard-test":
check_gtk()
from xpra.gtk_common import gtk_view_keyboard
return gtk_view_keyboard.main()
elif mode=="keymap":
check_gtk()
from xpra.gtk_common import keymap
return keymap.main()
elif mode=="gtk-info":
check_gtk()
from xpra.scripts import gtk_info
return gtk_info.main()
elif mode=="gui-info":
check_gtk()
from xpra.platform import gui
return gui.main()
elif mode=="network-info":
from xpra.net import net_util
return net_util.main()
elif mode=="compression":
from xpra.net import compression
return compression.main()
elif mode=="packet-encoding":
from xpra.net import packet_encoding
return packet_encoding.main()
elif mode=="path-info":
from xpra.platform import paths
return paths.main()
elif mode=="printing-info":
from xpra.platform import printing
return printing.main(args)
elif mode=="version-info":
from xpra.scripts import version
return version.main()
elif mode=="toolbox":
check_gtk()
from xpra.client.gtk_base import toolbox
return toolbox.main()
elif mode=="colors-test":
check_gtk()
from xpra.client.gtk_base.example import colors
return colors.main()
elif mode=="colors-gradient-test":
check_gtk()
from xpra.client.gtk_base.example import colors_gradient
return colors_gradient.main()
elif mode=="transparent-colors":
check_gtk()
from xpra.client.gtk_base.example import transparent_colors
return transparent_colors.main()
elif mode=="transparent-window":
check_gtk()
from xpra.client.gtk_base.example import transparent_window
return transparent_window.main()
elif mode == "initenv":
if not POSIX:
raise InitExit(EXIT_UNSUPPORTED, "initenv is not supported on this OS")
from xpra.server.server_util import xpra_runner_shell_script, write_runner_shell_scripts
script = xpra_runner_shell_script(script_file, os.getcwd(), options.socket_dir)
write_runner_shell_scripts(script, False)
return 0
elif mode=="auth":
return run_auth(options, args)
elif mode == "showconfig":
return run_showconfig(options, args)
elif mode == "showsetting":
return run_showsetting(args)
else:
from xpra.scripts.parsing import get_usage
if mode!="help":
print("Invalid subcommand '%s'" % (mode,))
print("Usage:")
from xpra.platform.features import LOCAL_SERVERS_SUPPORTED
if not LOCAL_SERVERS_SUPPORTED:
print("(this xpra installation does not support starting local servers)")
cmd = os.path.basename(sys.argv[0])
for x in get_usage():
print("\t%s %s" % (cmd, x))
print()
print("see 'man xpra' or 'xpra --help' for more details")
return 1
def find_session_by_name(opts, session_name):
from xpra.platform.paths import get_nodock_command
dotxpra = DotXpra(opts.socket_dir, opts.socket_dirs)
socket_paths = dotxpra.socket_paths(check_uid=getuid(), matching_state=DotXpra.LIVE)
if not socket_paths:
return None
id_sessions = {}
for socket_path in socket_paths:
cmd = get_nodock_command()+["id", "socket://%s" % socket_path]
proc = Popen(cmd, stdout=PIPE, stderr=PIPE)
id_sessions[socket_path] = proc
now = monotonic_time()
while any(proc.poll() is None for proc in id_sessions.values()) and monotonic_time()-now<10:
time.sleep(0.5)
session_uuid_to_path = {}
for socket_path, proc in id_sessions.items():
if proc.poll()==0:
out, err = proc.communicate()
d = {}
for line in bytestostr(out or err).splitlines():
try:
k,v = line.split("=", 1)
d[k] = v
except ValueError:
continue
name = d.get("session-name")
uuid = d.get("uuid")
if name==session_name and uuid:
session_uuid_to_path[uuid] = socket_path
if not session_uuid_to_path:
return None
if len(session_uuid_to_path)>1:
raise InitException("more than one session found matching '%s'" % session_name)
return "socket://%s" % tuple(session_uuid_to_path.values())[0]
def parse_ssh_string(ssh_setting):
ssh_cmd = shlex.split(ssh_setting, posix=not WIN32)
if ssh_cmd[0]=="auto":
#try paramiko:
try:
from xpra.net.ssh import nogssapi_context
with nogssapi_context():
import paramiko
assert paramiko
ssh_cmd = ["paramiko"]
if is_debug_enabled("ssh"):
Logger("ssh").info("using paramiko ssh backend")
except ImportError as e:
if is_debug_enabled("ssh"):
Logger("ssh").info("no paramiko: %s" % e)
from xpra.platform.features import DEFAULT_SSH_COMMAND
ssh_cmd = shlex.split(DEFAULT_SSH_COMMAND)
return ssh_cmd
def add_ssh_args(username, password, host, ssh_port, key, is_putty=False, is_paramiko=False):
args = []
if password and is_putty:
args += ["-pw", password]
if username and not is_paramiko:
args += ["-l", username]
if ssh_port and ssh_port!=22:
#grr why bother doing it different?
if is_putty:
args += ["-P", str(ssh_port)]
elif not is_paramiko:
args += ["-p", str(ssh_port)]
if not is_paramiko:
args += ["-T", host]
if key:
key_path = os.path.abspath(key)
if WIN32 and is_putty:
# tortoise plink works with either slash, backslash needs too much escaping
# because of the weird way it's passed through as a ProxyCommand
key_path = "\"" + key.replace("\\", "/") + "\"" # pragma: no cover
args += ["-i", key_path]
return args
def add_ssh_proxy_args(username, password, host, ssh_port, pkey, ssh, is_putty=False, is_paramiko=False):
args = []
proxyline = ssh
if is_putty:
proxyline += ["-nc", "%host:%port"]
elif not is_paramiko:
proxyline += ["-W", "%h:%p"]
# the double quotes are in case the password has something like "&"
proxyline += add_ssh_args(username, password, host, ssh_port, pkey, is_putty, is_paramiko)
if is_putty:
args += ["-proxycmd", " ".join(proxyline)]
elif not is_paramiko:
args += ["-o", "ProxyCommand " + " ".join(proxyline)]
return args
def parse_proxy_attributes(display_name):
import re
# Notes:
# (1) this regex permits a "?" in the password or username (because not just splitting at "?").
# It doesn't look for the next "?" until after the "@", where a "?" really indicates
# another field.
# (2) all characters including "@"s go to "userpass" until the *last* "@" after which it all goes
# to "hostport"
reout = re.search("\\?proxy=(?P<p>((?P<userpass>.+)@)?(?P<hostport>[^?]+))", display_name)
if not reout:
return display_name, {}
try:
desc_tmp = {}
# This one should *always* return a host, and should end with an optional numeric port
hostport = reout.group("hostport")
hostport_match = re.match(r"(?P<host>[^:]+)($|:(?P<port>\d+)$)", hostport)
if not hostport_match:
raise RuntimeError("bad format for 'hostport': '%s'" % hostport)
host = hostport_match.group("host")
if not host:
raise RuntimeError("bad format: missing host in '%s'" % hostport)
desc_tmp["proxy_host"] = host
if hostport_match.group("port"):
try:
desc_tmp["proxy_port"] = int(hostport_match.group("port"))
except ValueError:
raise RuntimeError("bad format: proxy port '%s' is not a number" % hostport_match.group("port")) from None
userpass = reout.group("userpass")
if userpass:
# The username ends at the first colon. This decision was not unique: I could have
# allowed one colon in username if there were two in the string.
userpass_match = re.match("(?P<username>[^:]+)(:(?P<password>.+))?", userpass)
if not userpass_match:
raise RuntimeError("bad format for 'userpass': '%s'" % userpass)
# If there is a "userpass" part, then it *must* have a username
username = userpass_match.group("username")
if not username:
raise RuntimeError("bad format: missing username in '%s'" % userpass)
desc_tmp["proxy_username"] = username
password = userpass_match.group("password")
if password:
desc_tmp["proxy_password"] = password
except RuntimeError:
sshlog = Logger("ssh")
sshlog.error("bad proxy argument: " + reout.group(0))
return display_name, {}
else:
# rip out the part we've processed
display_name = display_name[:reout.start()] + display_name[reout.end():]
return display_name, desc_tmp
def _sep_pos(display_name):
#split the display name on ":" or "/"
scpos = display_name.find(":")
slpos = display_name.find("/")
if scpos<0:
return slpos
elif slpos<0:
return scpos
return min(scpos, slpos)
def parse_display_name(error_cb, opts, display_name, session_name_lookup=False):
if WIN32:
from xpra.platform.win32.dotxpra import PIPE_PREFIX # pragma: no cover
else:
PIPE_PREFIX = None
if display_name.startswith("/") and POSIX:
display_name = "socket://"+display_name
desc = {"display_name" : display_name}
display_name, proxy_attrs = parse_proxy_attributes(display_name)
desc.update(proxy_attrs)
pos = _sep_pos(display_name)
if pos<0 or (display_name and display_name[0] in "0123456789"):
match = None
if POSIX:
#maybe this is just the display number without the ":" prefix?
try:
if pos>0:
display_name = ":%i" % int(display_name[:pos])
else:
display_name = ":%i" % int(display_name)
match = True
except ValueError:
pass
elif WIN32: # pragma: no cover
display_name = "named-pipe://%s%s" % (PIPE_PREFIX, display_name)
match = True
if session_name_lookup and not match:
#try to find a session whose "session-name" matches:
match = find_session_by_name(opts, display_name)
if match:
display_name = match
#display_name may have been updated, re-parse it:
pos = _sep_pos(display_name)
if pos<0:
error_cb("unknown format for display name: %s" % display_name)
protocol = display_name[:pos]
#the separator between the protocol and the rest can be ":", "/" or "://"
#but the separator value we use thereafter can only be ":" or "/"
#because we want strings like ssl://host:port/DISPLAY to be parsed into ["ssl", "host:port", "DISPLAY"]
psep = ""
if display_name[pos]==":":
psep += ":"
pos += 1
scount = 0
while display_name[pos]=="/" and scount<2:
psep += "/"
pos += 1
scount += 1
if protocol=="socket":
#socket paths may start with a slash!
#so socket:/path means that the slash is part of the path
if psep==":/":
psep = psep[:-1]
pos -= 1
if psep not in (":", "/", "://"):
error_cb("unknown format for protocol separator '%s' in display name: %s" % (psep, display_name))
afterproto = display_name[pos:] #ie: "host:port/DISPLAY"
separator = psep[-1] #ie: "/"
parts = afterproto.split(separator, 1) #ie: "host:port", "DISPLAY"
def parse_username_and_password(s):
ppos = s.find(":")
if ppos>=0:
password = s[ppos+1:]
username = s[:ppos]
else:
username = s
password = ""
#fugly: we override the command line option after parsing the string:
if username:
desc["username"] = username
opts.username = username
if password:
opts.password = password
desc["password"] = password
return username, password
def parse_host_string(host, default_port=DEFAULT_PORT):
"""
Parses [username[:password]@]host[:port]
and returns username, password, host, port
missing arguments will be empty (username and password) or 0 (port)
"""
upos = host.rfind("@")
username = None
password = None
port = default_port
if upos>=0:
#HOST=username@host
username, password = parse_username_and_password(host[:upos])
host = host[upos+1:]
port_str = None
if host.count(":")>=2:
#more than 2 ":", assume this is IPv6:
if host.startswith("["):
#if we have brackets, we can support: "[HOST]:SSHPORT"
epos = host.find("]")
if epos<0:
error_cb("invalid host format, expected IPv6 [..]")
port_str = host[epos+1:] #ie: ":22"
if port_str.startswith(":"):
port_str = port_str[1:] #ie: "22"
host = host[1:epos] #ie: "[HOST]"
else:
#ie: fe80::c1:ac45:7351:ea69%eth1:14500 -> ["fe80::c1:ac45:7351:ea69", "eth1:14500"]
devsep = host.split("%")
if len(devsep)==2:
parts = devsep[1].split(":", 1) #ie: "eth1:14500" -> ["eth1", "14500"]
if len(parts)==2:
host = "%s%%%s" % (devsep[0], parts[0])
port_str = parts[1] #ie: "14500"
else:
parts = host.split(":")
if len(parts[-1])>4:
port_str = parts[-1]
host = ":".join(parts[:-1])
else:
#otherwise, we have to assume they are all part of IPv6
#we could count them at split at 8, but that would be just too fugly
pass
elif host.find(":")>0:
host, port_str = host.split(":", 1)
if port_str:
try:
port = int(port_str)
except ValueError:
error_cb("invalid port number specified: %s" % port_str)
if port<=0 or port>=2**16:
error_cb("invalid port number: %s" % port)
desc["port"] = port
if host=="":
host = "127.0.0.1"
desc["host"] = host
desc["local"] = is_local(host)
return username, password, host, port
def parse_remote_display(s):
if not s:
return
qpos = s.find("?")
cpos = s.find(",")
display = None
options_str = None
if qpos>=0 and (qpos<cpos or cpos<0):
#query string format, ie: "DISPLAY?key1=value1&key2=value2#extra_stuff
attr_sep = "&"
parts = s.split("?", 1)
s = parts[0].split("#")[0]
options_str = parts[1]
elif cpos>0 and (cpos<qpos or qpos<0):
#csv string format,
# ie: DISPLAY,key1=value1,key2=value2
# or: key1=value1,key2=value2
attr_sep = ","
parts = s.split(",", 1)
if parts[0].find("=")>0:
#if the first part is a key=value,
#assume it is part of the parameters
parts = ["", s]
display = ""
if len(parts)==2:
options_str = parts[1]
elif s.find("=")>0:
#ie: just one key=value
#(so this is not a display)
display = ""
attr_sep = ","
parts = ["", s]
else:
parts = []
if display is None:
try:
assert [int(x) for x in s.split(".")] #ie: ":10.0" -> [10, 0]
display = ":" + s #ie: ":10.0"
except ValueError:
display = s #ie: "tcp://somehost:10000/"
desc["display"] = display
opts.display = display
desc["display_as_args"] = [display]
if options_str:
#parse extra attributes
d = parse_simple_dict(options_str, attr_sep)
for k,v in d.items():
if k in desc:
warn("Warning: cannot override '%s' with URI" % k)
else:
desc[k] = v
if protocol=="ssh":
desc.update({
"type" : "ssh",
"proxy_command" : ["_proxy"],
"exit_ssh" : opts.exit_ssh,
})
desc["display"] = None
desc["display_as_args"] = []
host = parts[0]
if len(parts)>1:
parse_remote_display(parts[1])
#ie: ssh=["/usr/bin/ssh", "-v"]
ssh = parse_ssh_string(opts.ssh)
full_ssh = ssh[:]
#maybe restrict to win32 only?
ssh_cmd = ssh[0].lower()
is_putty = ssh_cmd.endswith("plink") or ssh_cmd.endswith("plink.exe")
is_paramiko = ssh_cmd.split(":")[0]=="paramiko"
if is_paramiko:
ssh[0] = "paramiko"
desc["is_paramiko"] = is_paramiko
if opts.ssh.find(":")>0: