[00:02.180 --> 00:05.380]  So, hello everyone. Thank you for tuning in.
[00:05.460 --> 00:07.680]  Today, we're going to talk about Apache Spark.
[00:07.840 --> 00:09.680]  Now, I will just assume that while some of you
[00:09.680 --> 00:11.600]  may be familiar with the technology,
[00:11.700 --> 00:13.840]  a lot of you did not have actual chance
[00:13.840 --> 00:14.820]  to play with the cluster.
[00:14.820 --> 00:17.040]  So, we'll take it right from the beginning.
[00:17.840 --> 00:19.560]  So yeah, we'll see how Spark works,
[00:19.560 --> 00:21.280]  what makes it so popular,
[00:21.280 --> 00:23.980]  how to view some common and default settings,
[00:23.980 --> 00:25.860]  and obviously, some bones along the way
[00:25.860 --> 00:29.220]  to bypass authentication and execute code, right?
[00:29.360 --> 00:30.680]  So, let's get going, right?
[00:30.680 --> 00:34.120]  So, our story takes us back to 2008,
[00:34.120 --> 00:37.460]  when the economy was not much better than today's,
[00:37.460 --> 00:39.720]  but Internet Explorer was still a thing,
[00:39.720 --> 00:42.300]  and the big tech companies were doing what they did best,
[00:42.300 --> 00:44.480]  quietly collecting the world's data.
[00:44.560 --> 00:46.040]  When you gather that much data,
[00:46.040 --> 00:49.160]  and I'm talking about terabytes and terabytes of data,
[00:49.160 --> 00:50.920]  performing even the simplest calculations
[00:50.920 --> 00:52.860]  becomes quite challenging.
[00:53.020 --> 00:54.580]  Now, you can either get a mainframe for half a million
[00:54.580 --> 00:57.060]  dollars, plus whatever it takes to get an intern
[00:57.060 --> 01:00.300]  to work on it, or you can actually distribute
[01:00.300 --> 01:03.300]  your files over thousands and thousands of servers.
[01:03.520 --> 01:07.600]  That way, your computations are small, though multiplied,
[01:07.600 --> 01:10.360]  and that's one way to handle the load.
[01:10.360 --> 01:13.240]  And the standard way of doing so, I would say,
[01:13.240 --> 01:15.980]  is by using the Apache Hadoop framework.
[01:16.420 --> 01:18.540]  Now, if you've ever worked with Hadoop,
[01:18.540 --> 01:22.020]  you must know that it's actually quite a hostile environment.
[01:22.140 --> 01:24.200]  It's a very complex environment.
[01:24.300 --> 01:26.680]  So, first of all, just to give you a quick example,
[01:26.680 --> 01:30.000]  you need to fragment your files using the HDFS file system,
[01:30.000 --> 01:34.500]  so you end up with multiple fragments on multiple servers,
[01:34.500 --> 01:37.180]  these files, and then you need to do coherent computations
[01:37.180 --> 01:38.540]  on these small fragments,
[01:38.540 --> 01:40.280]  so you need the MapReduce framework.
[01:40.340 --> 01:42.460]  But then you end up with thousands of processes
[01:42.460 --> 01:44.100]  running on multiple machines,
[01:44.100 --> 01:46.320]  so you need a way to schedule them, enter YARN,
[01:46.320 --> 01:48.080]  but then you want to do some SQL querying,
[01:48.080 --> 01:50.640]  so you need Apache Drill, et cetera, et cetera.
[01:50.640 --> 01:53.160]  So, you end up with this plethora of tools,
[01:53.160 --> 01:54.280]  each with their own latencies,
[01:54.800 --> 01:58.400]  with their own learning curve, et cetera,
[01:58.400 --> 02:00.560]  just to cover all your needs in data analytics.
[02:00.560 --> 02:02.620]  So, just to give you an example,
[02:02.620 --> 02:06.020]  this is what it takes to do a simple word count on MapReduce.
[02:06.020 --> 02:07.420]  So, this is the first page of Java.
[02:07.420 --> 02:09.800]  This is the second page, which is downright outrageous
[02:09.800 --> 02:12.520]  for a data processing solution, right?
[02:13.360 --> 02:16.340]  So, in our story, we're around 2009,
[02:16.340 --> 02:19.680]  and there's this guy named Matei Azaria,
[02:19.680 --> 02:21.940]  who was at Berkeley, I believe, at the time,
[02:22.400 --> 02:24.920]  who I imagine looked at this code and went,
[02:24.920 --> 02:25.880]  are you fucking crazy?
[02:25.880 --> 02:28.340]  That's not going to work.
[02:28.560 --> 02:30.600]  A word count is a simple operation.
[02:30.600 --> 02:34.220]  A word count should be done in five lines of code.
[02:34.500 --> 02:35.960]  Apache Spark was born.
[02:36.360 --> 02:37.620]  Now, this is not the official story.
[02:37.620 --> 02:38.880]  The official story is that he was working
[02:38.880 --> 02:41.140]  on something called Mesos, which is a resource manager,
[02:41.140 --> 02:43.000]  and he invented Spark as a proof of concept,
[02:43.000 --> 02:45.160]  but I like my version better.
[02:45.340 --> 02:46.860]  So, anyway, needless to say,
[02:46.860 --> 02:48.780]  we'll go through the code later on,
[02:48.780 --> 02:50.240]  but right off the bat, you can see
[02:50.240 --> 02:52.380]  that you can do pretty much the same processing
[02:52.380 --> 02:53.840]  that you would do in Apache Hadoop,
[02:53.840 --> 02:57.320]  except that you can do it at 10 times less code than Spark.
[02:57.560 --> 02:59.820]  But it's also much faster.
[03:00.280 --> 03:02.840]  So, it's three times faster on 10 times less nodes,
[03:02.840 --> 03:04.340]  so it's 30 times faster.
[03:04.340 --> 03:06.340]  You see, Hadoop will flush everything to disk
[03:06.340 --> 03:07.660]  whenever it gets a chance,
[03:07.660 --> 03:10.300]  whereas Spark will try to keep everything in memory.
[03:10.300 --> 03:11.440]  It was developed in an era
[03:11.440 --> 03:13.160]  where memory was getting cheaper and cheaper,
[03:13.160 --> 03:14.560]  so it kind of makes sense.
[03:14.560 --> 03:16.200]  But if you're running in a cloud environment
[03:16.200 --> 03:18.680]  where you pay for the usage,
[03:19.820 --> 03:21.640]  having something that runs 30 times faster
[03:21.640 --> 03:24.540]  means that it's 30 times cheaper, actually.
[03:24.900 --> 03:28.860]  And that partly explains the gigantic boom
[03:28.860 --> 03:33.320]  and popularity of Spark amongst the big data companies.
[03:33.320 --> 03:36.220]  If you're serious about your big data processing
[03:36.220 --> 03:38.900]  and cluster front, you will be using Spark
[03:38.900 --> 03:41.180]  for at least some part of it.
[03:41.500 --> 03:42.780]  And so, yeah, Spark boomed,
[03:43.430 --> 03:45.240]  was adopted by the Apache Foundation,
[03:45.240 --> 03:49.500]  and it grew to encompass many aspects of data analytics.
[03:49.500 --> 03:50.880]  If you're gonna do SQL, streaming,
[03:50.880 --> 03:52.460]  machine learning, graph processing,
[03:52.460 --> 03:53.860]  you know, it got you covered.
[03:54.180 --> 03:56.440]  So it's developed half in Java, half in Scala,
[03:56.440 --> 03:59.320]  but it also has some connectors in Python and R.
[03:59.320 --> 04:01.580]  So, yeah, awesome.
[04:01.640 --> 04:06.320]  So basically, when you have this powerful framework,
[04:06.320 --> 04:08.360]  suddenly, you know, everything looks like a nail.
[04:08.360 --> 04:09.980]  So if you're gonna do revenue prediction,
[04:09.980 --> 04:11.900]  you just hook Spark to your, I don't know,
[04:11.900 --> 04:14.420]  Redshift cluster, download those financial statements,
[04:14.420 --> 04:16.280]  and then, you know, you can do yourself.
[04:16.280 --> 04:18.640]  If you're gonna do fraud analysis, user targeting,
[04:18.640 --> 04:20.660]  same, you hook it up to your Cassandra database,
[04:20.660 --> 04:22.360]  your Astra buckets, you, you know,
[04:22.360 --> 04:24.680]  upload or download those log files,
[04:24.680 --> 04:26.560]  and then you do your computation.
[04:26.560 --> 04:28.680]  So basically, what I'm saying is that
[04:29.180 --> 04:31.680]  Apache Spark is often in the,
[04:31.680 --> 04:34.320]  at the junction of almost every important data store
[04:34.320 --> 04:35.460]  inside the company.
[04:35.700 --> 04:38.240]  And that's what made it interesting, at least for me,
[04:38.240 --> 04:41.480]  because, well, if you can access this tool,
[04:41.480 --> 04:43.360]  you have pretty much,
[04:43.360 --> 04:46.480]  you're pretty much open to every database that's out there.
[04:46.480 --> 04:48.380]  So, you know, how is it protected?
[04:48.380 --> 04:49.240]  How does it work?
[04:49.240 --> 04:50.920]  What is its security, et cetera?
[04:51.020 --> 04:52.720]  And so what I did, I did what I,
[04:52.720 --> 04:54.080]  any sensible person would do,
[04:54.080 --> 04:55.980]  I went to the documentation, to their website,
[04:55.980 --> 04:57.540]  and I went to the security page,
[04:57.540 --> 05:01.740]  and lo and behold, from this magnificent sentence
[05:01.740 --> 05:05.120]  that says security in Spark is off by default, well, fuck.
[05:06.580 --> 05:07.800]  And what does it do again?
[05:07.800 --> 05:11.020]  Oh yeah, it's this gigantic framework cluster
[05:11.020 --> 05:13.200]  of hundreds and hundreds of machines
[05:13.200 --> 05:16.420]  that have access to every database inside the company.
[05:16.420 --> 05:17.300]  And what does it do?
[05:17.300 --> 05:19.720]  Oh yeah, it does distributed data processing,
[05:19.720 --> 05:21.700]  or as I'd like to call it,
[05:21.700 --> 05:24.120]  it's really distributed remote code execution.
[05:24.120 --> 05:25.660]  So I thought this would be fun.
[05:26.120 --> 05:27.880]  And I hope I piqued your interest
[05:27.880 --> 05:29.780]  because really that's what sparked my,
[05:29.780 --> 05:32.300]  the whole endeavor, no pun intended.
[05:32.620 --> 05:34.440]  And I hope I got you interested enough
[05:34.440 --> 05:37.080]  to actually continue along this presentation
[05:37.080 --> 05:38.540]  so we can explore how it works
[05:38.540 --> 05:43.220]  and some bones that I found, right?
[05:43.220 --> 05:46.240]  So, oh yeah, in keeping with DEF CON's tradition,
[05:46.240 --> 05:49.380]  I think it's time for drinking.
[05:50.500 --> 05:54.240]  I should not do this online, but oh, what the heck?
[05:55.260 --> 05:57.760]  So to DEF CON safe mode, cheers.
[06:01.360 --> 06:02.920]  Yeah, that will kick in later.
[06:02.920 --> 06:06.880]  Anyway, so what is Spark and how does it work?
[06:07.320 --> 06:08.620]  I hope it's going to be one take
[06:08.620 --> 06:10.380]  because I cannot do that five times.
[06:10.460 --> 06:12.000]  So how does it work?
[06:12.000 --> 06:14.400]  So as in everything in distributed processing,
[06:14.400 --> 06:16.620]  basically we start with a bunch of machines
[06:16.620 --> 06:18.220]  that we're going to call workers.
[06:18.860 --> 06:22.980]  Now these workers are just basically brainless machines
[06:22.980 --> 06:24.880]  that will execute whatever is sent their way.
[06:24.880 --> 06:28.040]  Now on each worker, you find a JVM process,
[06:28.040 --> 06:31.300]  so a Spark process that's called an executor
[06:31.300 --> 06:33.940]  and that's going to do the actual execution.
[06:34.440 --> 06:36.700]  Now you can have many executors on each worker
[06:37.180 --> 06:38.980]  and it's the basically executor
[06:38.980 --> 06:42.040]  that defines your power in terms of parallelism.
[06:42.040 --> 06:44.160]  So if you have three, you have 10 workers
[06:44.160 --> 06:46.200]  on each worker or each machine,
[06:46.200 --> 06:47.500]  basically you have three executors
[06:47.500 --> 06:49.740]  so you can run 30 tasks in parallel.
[06:51.080 --> 06:53.520]  So yeah, you have like some default HTTP port
[06:53.520 --> 06:55.520]  that gives a status of what's going on
[06:55.520 --> 06:57.380]  on the machine you're not interested in,
[06:57.380 --> 06:59.220]  but you have the random RPC port
[06:59.220 --> 07:01.420]  that is set for each executor.
[07:01.420 --> 07:04.480]  And it's this one that the rest of the cluster contacts
[07:04.480 --> 07:09.080]  in order to send and receive a status from the worker.
[07:10.020 --> 07:12.260]  Now the worker is fine, but the most important piece
[07:12.260 --> 07:13.660]  or one of the most important pieces,
[07:13.660 --> 07:15.920]  the cluster manager, and that's our second piece.
[07:16.500 --> 07:18.240]  And the cluster manager's sole job
[07:18.240 --> 07:20.660]  is to actually schedule the application.
[07:20.660 --> 07:23.020]  So the work that is coming and say,
[07:23.020 --> 07:24.640]  oh yeah, this application is going to run
[07:24.640 --> 07:26.420]  on these two workers, that one is going to run
[07:26.420 --> 07:27.440]  on the other three workers.
[07:27.440 --> 07:31.180]  So it has a status of the workers, what they're doing,
[07:31.180 --> 07:34.340]  are they online or not, are they busy or not, et cetera.
[07:34.520 --> 07:36.880]  Every component inside Spark communicates
[07:36.880 --> 07:38.560]  using this Spark RPC protocol
[07:38.560 --> 07:41.040]  that we will detail a little bit later on.
[07:41.040 --> 07:42.400]  And just to show you what, you know,
[07:42.400 --> 07:45.160]  to demystify this Spark cluster,
[07:45.160 --> 07:46.460]  this is what it looks like.
[07:46.880 --> 07:50.600]  This is the basic UI of the Spark master on port 8080.
[07:50.600 --> 07:52.060]  Oh, sorry about that.
[07:52.160 --> 07:54.040]  So yeah, you can see that we have like, you know,
[07:54.040 --> 07:55.820]  two workers, one which is dead.
[07:55.820 --> 07:58.420]  We have an application that's running on two cores.
[07:58.420 --> 08:00.660]  And since each worker has one core,
[08:00.660 --> 08:02.460]  so this application is running on two cores.
[08:02.460 --> 08:04.840]  It's being distributed over two machines.
[08:05.640 --> 08:10.140]  But the most important port really is port 7077,
[08:10.140 --> 08:12.120]  because that's the one that we're going to contact
[08:12.120 --> 08:14.500]  in order to send an application, register it,
[08:14.500 --> 08:17.260]  and schedule it to be, you know,
[08:17.260 --> 08:20.120]  executed in parallel on these workers.
[08:20.640 --> 08:22.400]  And it's going to be these basic hypotheses
[08:22.400 --> 08:24.840]  that we're going to make along this presentation
[08:24.840 --> 08:29.040]  that we have access to that, to this specific ports.
[08:29.360 --> 08:31.980]  And the third component, and indeed the most component,
[08:31.980 --> 08:33.860]  the most important, sorry, is the driver.
[08:33.860 --> 08:35.860]  And the driver is really the brain
[08:35.860 --> 08:38.140]  behind the application orchestration.
[08:38.140 --> 08:39.860]  It's the one that's going to slice up your application
[08:39.860 --> 08:42.640]  and small tasks, contact the cluster manager,
[08:42.640 --> 08:44.320]  receive the worker, and then send,
[08:44.320 --> 08:46.280]  distribute your workload to these worker
[08:46.280 --> 08:49.820]  and, you know, make sure that everything is going smoothly
[08:49.820 --> 08:51.420]  and aggregate the results.
[08:52.500 --> 08:55.420]  And the driver in a typical case or typical scenario
[08:55.420 --> 08:56.840]  is going to be running on your laptop.
[08:56.840 --> 08:58.660]  So you enter a network,
[08:58.660 --> 09:00.380]  you have the cluster manager running somewhere,
[09:00.380 --> 09:02.560]  you have the workers running somewhere else,
[09:02.560 --> 09:04.340]  and you're going to boot up the driver,
[09:04.340 --> 09:07.380]  write your application in Java, Scala, Python, or whatever,
[09:07.380 --> 09:08.340]  boot up the driver,
[09:08.340 --> 09:10.040]  and then send it to the cluster manager
[09:10.040 --> 09:11.980]  to be scheduled on the workers,
[09:11.980 --> 09:13.520]  hopefully to execute some code.
[09:14.080 --> 09:18.300]  So that's one setup that's kind of common.
[09:18.900 --> 09:20.620]  Right, so recon.
[09:20.620 --> 09:23.540]  So let's take this specific scenario that I just listed.
[09:23.540 --> 09:24.900]  Let's say you end up in a network
[09:25.360 --> 09:27.500]  and you want to find, you want to hunt for some spark
[09:27.500 --> 09:28.960]  in order to exploit it.
[09:29.640 --> 09:31.240]  How would you go about it?
[09:31.240 --> 09:33.740]  Well, the first thing you want to do, I guess,
[09:33.740 --> 09:37.440]  is basically if you're in an AWS environment,
[09:37.440 --> 09:39.040]  you can simply use the API,
[09:39.040 --> 09:41.120]  if you have the proper access rights, obviously,
[09:41.120 --> 09:44.460]  to look for every machine that has spark, master,
[09:44.460 --> 09:46.280]  or other keywords in their labels.
[09:46.280 --> 09:48.600]  Similarly, if you're in a Kube environment, Kubernetes,
[09:48.600 --> 09:49.720]  it's the same thing.
[09:49.720 --> 09:51.700]  But what if you end up in a more traditional network
[09:51.700 --> 09:53.900]  where you have to nmap the shit out of things?
[09:53.900 --> 09:56.020]  How can you find that cluster?
[09:56.080 --> 09:58.920]  Because you see nmap does not really support spark.
[09:58.920 --> 10:01.420]  So we're in a kind of a pickle here.
[10:01.420 --> 10:02.720]  So the first thing we want to do
[10:03.220 --> 10:06.360]  is be able to fingerprint spark properly using nmap.
[10:06.360 --> 10:07.280]  And to do that,
[10:07.280 --> 10:10.020]  I simply whip out a spark cluster in my lab.
[10:10.020 --> 10:12.460]  I put some Wireshark in the middle,
[10:12.460 --> 10:14.100]  and I analyze the traffic.
[10:14.100 --> 10:14.800]  And if you do that,
[10:14.800 --> 10:16.680]  this is the blob that you're going to see.
[10:19.240 --> 10:23.000]  And basically, the way to decompose this is fairly simple,
[10:23.000 --> 10:24.240]  and it's quite repetitive, actually,
[10:24.240 --> 10:25.800]  once you get to know Spark RPC.
[10:25.800 --> 10:27.880]  So this is the Spark RPC I was talking about.
[10:28.100 --> 10:31.860]  It always starts with the same thing, the same header.
[10:31.860 --> 10:33.360]  It's going to be, most of the time,
[10:33.360 --> 10:36.640]  21 bytes of data followed by the payload, right?
[10:36.640 --> 10:39.120]  And these 21 bytes of data,
[10:39.500 --> 10:42.720]  a common header when you send Spark RPC commands,
[10:42.720 --> 10:45.080]  is composed of seven bytes of null bytes,
[10:45.080 --> 10:46.620]  followed by two magic bytes.
[10:46.620 --> 10:47.560]  I call them magic bytes,
[10:47.560 --> 10:49.940]  but basically they depend on the RPC endpoint
[10:49.940 --> 10:50.960]  that you're trying to reach
[10:50.960 --> 10:52.840]  and the type of message that you're sending.
[10:52.900 --> 10:54.200]  So in this case,
[10:54.200 --> 10:58.440]  I chose the example of verify endpoint,
[10:58.440 --> 11:00.700]  check existence message.
[11:01.260 --> 11:04.180]  And its magic byte is C305.
[11:04.180 --> 11:05.200]  If it's, I don't know,
[11:05.200 --> 11:07.620]  if it's a worker sending a heartbeat message
[11:07.620 --> 11:08.600]  to the cluster manager,
[11:08.600 --> 11:11.620]  it's going to be 2B0F or whatever.
[11:11.780 --> 11:13.600]  But in this case, it's C305.
[11:13.600 --> 11:14.820]  But anyway, it doesn't matter.
[11:15.300 --> 11:18.280]  Next, we have eight bytes of data, like random data.
[11:18.280 --> 11:19.760]  You can put anything you want here.
[11:19.760 --> 11:23.100]  They'll just be echoed back by the cluster manager.
[11:23.420 --> 11:26.920]  And finally, we have four bytes of the size of the data
[11:26.920 --> 11:27.700]  that's going to follow,
[11:27.700 --> 11:29.560]  so the payload that will follow.
[11:29.560 --> 11:31.120]  What is this payload?
[11:31.320 --> 11:33.140]  Well, you have a bunch of data IP addresses
[11:33.140 --> 11:34.240]  that we don't care about,
[11:34.240 --> 11:35.780]  but the most important thing really,
[11:35.780 --> 11:37.960]  the heart of the payload, if you will,
[11:37.960 --> 11:40.440]  is this serialized object that you can see here.
[11:40.440 --> 11:43.800]  So it's a serialized JVM object called check existence.
[11:43.800 --> 11:44.960]  And if you look at the source code,
[11:44.960 --> 11:46.920]  check existence is actually a class.
[11:46.920 --> 11:48.660]  Well, it's a Scala case class,
[11:48.660 --> 11:51.840]  but just think of it as a regular class, which is immutable.
[11:52.220 --> 11:54.600]  And it has some interesting other properties in Scala
[11:54.600 --> 11:56.920]  that we simply don't give a fuck about,
[11:56.920 --> 11:58.420]  but yeah, it's a class.
[11:58.960 --> 12:02.760]  So check existence is a class and it has an attribute name.
[12:02.760 --> 12:04.500]  So basically the client or driver
[12:04.500 --> 12:07.320]  is sending this serialized class to the cluster manager
[12:07.320 --> 12:10.440]  to see if there's an endpoint bearing the name
[12:10.880 --> 12:11.600]  that it sent.
[12:11.600 --> 12:13.300]  And the name that it sent is obviously master.
[12:13.300 --> 12:17.500]  So it wants to know if there is a master RPC endpoint
[12:18.020 --> 12:20.800]  listening on the other end of the communication.
[12:21.080 --> 12:24.580]  And if that is the case, the cluster manager,
[12:24.580 --> 12:27.000]  well, will respond first with the 21 bytes of data
[12:27.000 --> 12:29.540]  that you can see it echoed back what we said before, right?
[12:29.540 --> 12:32.700]  And a different magic RPC, magic two bytes.
[12:32.860 --> 12:35.760]  But anyway, and if there is indeed an endpoint,
[12:35.760 --> 12:38.720]  it sends a Boolean that is set to true.
[12:38.740 --> 12:40.920]  So obviously I took this special exchange
[12:40.920 --> 12:43.180]  because you can use it and map to fingerprint
[12:43.180 --> 12:45.580]  the cluster manager, but you have to imagine
[12:45.580 --> 12:50.600]  that all Spark RPCs are these exchange of serialized data
[12:50.600 --> 12:53.360]  that are prefixed with some type of header,
[12:53.360 --> 12:55.740]  either a 21 bytes header or a 13 byte header
[12:55.740 --> 12:57.740]  that we will see a bit later on.
[12:57.740 --> 13:01.480]  But this is the main way that Spark components
[13:01.480 --> 13:03.700]  communicate with each other, just serialized objects.
[13:04.740 --> 13:07.180]  So once we have this exchange all mapped out,
[13:07.180 --> 13:10.420]  then we can use it and write a Lua script
[13:10.420 --> 13:12.540]  that will replicate the same thing.
[13:12.540 --> 13:13.980]  Once you do that, obviously,
[13:13.980 --> 13:15.720]  once you deconstruct it like this,
[13:15.720 --> 13:17.080]  it becomes quite obvious to do it.
[13:17.080 --> 13:19.660]  So I wrote an Nmap script that's on my repo.
[13:20.180 --> 13:22.580]  Sorry, and I will push it to the real Nmap repo
[13:22.580 --> 13:26.100]  once people have used it enough and debugged it enough.
[13:26.660 --> 13:28.980]  So yeah, now we are able to identify a Spark cluster.
[13:28.980 --> 13:31.100]  And you can see they've been true to their promise.
[13:31.100 --> 13:33.900]  Authentication is indeed disabled by default.
[13:33.900 --> 13:35.260]  So this is pretty great.
[13:35.980 --> 13:37.140]  Now comes the interesting part.
[13:37.140 --> 13:40.820]  Now we know how to locate Spark clusters inside a network.
[13:40.820 --> 13:43.700]  How can we execute code, right?
[13:43.700 --> 13:45.520]  Because that's what's really interesting.
[13:45.920 --> 13:48.480]  And now the standard way of submitting an application
[13:48.480 --> 13:51.360]  on to a Spark cluster is basically
[13:51.360 --> 13:53.500]  you write your application in Java or Scala,
[13:53.500 --> 13:57.340]  what have you, you package it as a JAR file,
[13:57.340 --> 13:58.500]  you boot up the driver,
[13:58.500 --> 14:01.500]  and then you send that JAR file to be executed
[14:02.540 --> 14:03.960]  or parsed, if you will,
[14:03.960 --> 14:05.800]  by the cluster manager and the workers.
[14:06.440 --> 14:08.680]  Now, the problem is this,
[14:08.680 --> 14:12.640]  I find writing Java boring, sorry,
[14:12.640 --> 14:15.400]  and Scala makes me want to shoot myself.
[14:15.400 --> 14:18.100]  So I decided to do it in Python.
[14:18.640 --> 14:20.820]  And luckily, Spark supports Python.
[14:20.820 --> 14:23.980]  In fact, they have an official wrapper called PySpark,
[14:24.820 --> 14:26.280]  which actually takes care
[14:26.280 --> 14:28.160]  of all the nasty stuff that I just described.
[14:28.160 --> 14:29.840]  So it will boot up the driver for you.
[14:29.840 --> 14:33.280]  It will serialize JVM object into Python object
[14:33.280 --> 14:34.600]  and vice versa and send it.
[14:34.600 --> 14:36.620]  And it will take care of all the nasty bits for you.
[14:36.620 --> 14:39.100]  So you just write a simple Python.
[14:40.260 --> 14:41.420]  And so, yeah, PySpark.
[14:41.420 --> 14:42.820]  And if you pip install PySpark,
[14:42.820 --> 14:44.340]  it really is just a thin wrapper.
[14:44.340 --> 14:46.580]  It downloads 200 megabytes of JAR file behind.
[14:46.580 --> 14:47.900]  So it's hilarious.
[14:47.900 --> 14:49.560]  Anyway, so from PySpark,
[14:49.560 --> 14:51.380]  we import these two classes,
[14:51.380 --> 14:52.780]  SparkContext and SparkConfiguration.
[14:52.780 --> 14:55.900]  It's very straightforward, very intuitive, I would say.
[14:55.900 --> 14:57.140]  So you define the configuration.
[14:57.140 --> 14:59.200]  We're going to name our application WordCount.
[14:59.200 --> 15:02.700]  We're going to point to the cluster, port 7077, obviously.
[15:02.820 --> 15:05.820]  Now you have to define your public IP
[15:05.820 --> 15:07.220]  that you're going to bind on.
[15:07.220 --> 15:08.460]  We're going to see why later.
[15:08.460 --> 15:11.380]  But if you forget this line, it will not work.
[15:11.400 --> 15:12.820]  No word does it say so in the doc,
[15:12.820 --> 15:14.620]  but if you forget this line, it will not work.
[15:14.660 --> 15:17.880]  So anyway, so you define the SparkContext.
[15:20.300 --> 15:21.840]  SparkContext is going to be the client
[15:21.840 --> 15:24.700]  that's going to initiate the communication,
[15:24.700 --> 15:26.700]  handle the communication with the cluster manager.
[15:26.700 --> 15:29.060]  And basically, when the Python interpreter
[15:29.060 --> 15:32.760]  comes to this line, it will boot up the JVM process.
[15:32.760 --> 15:33.860]  We talked about the driver,
[15:33.860 --> 15:35.060]  and it will initiate communication
[15:35.060 --> 15:36.340]  with the cluster manager.
[15:36.360 --> 15:39.440]  So quite naively and quite intuitively, I would say,
[15:39.440 --> 15:40.940]  what we want to do next is obviously
[15:42.140 --> 15:44.380]  just send the code that we want to...
[15:45.120 --> 15:46.400]  we want to be serialized
[15:46.400 --> 15:48.120]  and we want to be executed on the worker.
[15:48.120 --> 15:51.660]  So from subprocess P open the command ID.
[15:52.540 --> 15:53.640]  And if you do that,
[15:53.640 --> 15:55.900]  and this is really the first thing that I did.
[15:56.020 --> 15:57.980]  If you do that, it actually works.
[15:57.980 --> 16:01.880]  So I got this result back and I was like, well, hang on.
[16:02.160 --> 16:04.080]  I know that user, that's me.
[16:04.080 --> 16:07.240]  I have that user defined on my local machine,
[16:07.240 --> 16:09.300]  not on the worker or the cluster machines.
[16:09.300 --> 16:10.280]  What the fuck happened?
[16:10.280 --> 16:12.240]  So it turns out I actually just executed the code
[16:12.240 --> 16:13.200]  on my own machine.
[16:13.560 --> 16:15.520]  So if you look at Wireshark, what happened,
[16:15.520 --> 16:17.180]  because I was curious, what the fuck happened?
[16:17.180 --> 16:18.980]  I thought this stuff was supposed to serialize
[16:18.980 --> 16:21.280]  and send the command to be executed on the workers.
[16:21.280 --> 16:25.180]  Well, what happened was first we actually sent this,
[16:25.180 --> 16:27.420]  you know, 21 bytes of data, check existence class,
[16:27.420 --> 16:28.920]  yada, yada, everything's all right.
[16:28.920 --> 16:31.100]  There is indeed a cluster manager listening.
[16:31.420 --> 16:34.160]  Then we send the register application class,
[16:34.160 --> 16:36.420]  which contains all the information of the application
[16:36.420 --> 16:39.000]  that we can... that we're going to run.
[16:39.000 --> 16:41.180]  So you can see that we send, you know,
[16:41.180 --> 16:44.660]  word count plus another other properties, right?
[16:44.660 --> 16:48.400]  The cluster manager responded with executed added class.
[16:48.580 --> 16:51.680]  So they gave us actually two workers.
[16:51.680 --> 16:52.760]  So this is great.
[16:52.760 --> 16:53.680]  What happened next?
[16:53.680 --> 16:56.000]  Well, the driver, so our instance of PySpark
[16:56.000 --> 16:57.360]  simply unregistered the application
[16:57.360 --> 16:59.060]  and called off the whole thing.
[16:59.060 --> 17:00.340]  So what the hell?
[17:00.520 --> 17:02.660]  Well, to understand what happened actually,
[17:02.660 --> 17:05.480]  remember when I said that Spark contacts,
[17:05.480 --> 17:07.220]  PySpark will simply serialize whatever comes
[17:07.220 --> 17:08.940]  after the Spark contacts line?
[17:08.940 --> 17:10.140]  Yeah, I lied.
[17:10.660 --> 17:13.480]  Well, the thing is, in order to understand what happened,
[17:13.480 --> 17:15.720]  we basically need to talk about two concepts.
[17:15.720 --> 17:17.120]  First of all, Spark APIs,
[17:17.120 --> 17:18.960]  and second, the notion of a DAG
[17:18.960 --> 17:20.680]  and lazy evaluation in Spark.
[17:20.860 --> 17:25.060]  So Spark APIs, counter-intuitively I would say,
[17:25.060 --> 17:27.480]  Spark APIs are not methods, they're data structures.
[17:28.160 --> 17:30.280]  So let's say you want to open a file on Python
[17:30.280 --> 17:32.180]  or Java or Scala or whatever.
[17:32.240 --> 17:34.860]  You point to a file and then you call the method,
[17:34.860 --> 17:35.740]  the open method.
[17:35.740 --> 17:38.400]  This will only give you an object, a Python object,
[17:38.400 --> 17:40.320]  that refers to a handling memory, whatever,
[17:40.860 --> 17:43.040]  but a single threaded object nonetheless.
[17:43.740 --> 17:45.560]  If there is no parallelism there,
[17:46.140 --> 17:48.040]  and the proper way to do it in Spark
[17:48.040 --> 17:49.580]  and still get that parallelism
[17:49.580 --> 17:51.820]  is actually to call text file method
[17:51.820 --> 17:53.980]  of the Spark contacts class.
[17:53.980 --> 17:56.960]  So here SC is an instance of Spark contacts.
[17:56.960 --> 17:59.500]  So you point to a file, you call the text file method
[17:59.500 --> 18:02.000]  and text file method will grab that file,
[18:02.000 --> 18:03.100]  slice it into fragments
[18:03.100 --> 18:06.260]  and then load those two fragments in memory.
[18:06.920 --> 18:09.360]  Similarly, if you want to load a list,
[18:09.360 --> 18:11.220]  just don't define a list like that.
[18:11.220 --> 18:13.640]  This will give you only a single threaded object in memory.
[18:13.640 --> 18:15.960]  No, you need to call the parallelize method,
[18:15.960 --> 18:18.140]  which will take that list, slice it in half
[18:18.140 --> 18:21.540]  or three or four, whatever Spark seems necessary.
[18:21.740 --> 18:24.240]  And that way you will end up with two fragments in file
[18:24.240 --> 18:27.300]  on which you can perform work in a parallel fashion.
[18:27.540 --> 18:30.080]  Now, these fragments are called partitions.
[18:30.140 --> 18:31.800]  And this is a very frequent term in Spark
[18:31.800 --> 18:33.040]  that will come up a lot.
[18:33.040 --> 18:37.180]  Partitions are the main unit of work inside Spark.
[18:37.680 --> 18:40.820]  And these collection of partitions
[18:40.820 --> 18:43.040]  are called resilient distributed datasets,
[18:43.040 --> 18:45.340]  which is really the most fundamental API in Spark.
[18:45.340 --> 18:47.360]  There are others, but they're ultimately all based
[18:47.360 --> 18:50.300]  on RDDs at the core.
[18:50.500 --> 18:53.680]  So once we have these partition,
[18:53.680 --> 18:56.300]  then we can apply parallelized work
[18:56.300 --> 18:58.680]  using something called transformations in action.
[18:58.680 --> 19:00.000]  Now, what's a transformation?
[19:00.340 --> 19:01.820]  A transformation...
[19:01.820 --> 19:03.280]  Now, let's say I want to multiply every element
[19:03.280 --> 19:05.120]  on the list by 20, let's say.
[19:05.120 --> 19:07.600]  A concrete example will actually make it better.
[19:07.780 --> 19:09.720]  So I define a method called multiply,
[19:09.720 --> 19:13.460]  which takes an element, multiplies it by 20.
[19:13.580 --> 19:15.800]  And then I just call the map transformation,
[19:15.800 --> 19:19.460]  which will iterate over every element in every partition
[19:19.460 --> 19:21.940]  and yield the same number of elements
[19:21.940 --> 19:23.380]  in the same number of partitions.
[19:23.380 --> 19:24.620]  So that's a transformation.
[19:24.620 --> 19:26.140]  There are other types of transformation,
[19:26.140 --> 19:28.580]  which may or may not yield the same number of partitions
[19:29.380 --> 19:30.960]  and elements, by the way.
[19:30.960 --> 19:34.320]  So yeah, each Spark API basically defines
[19:34.320 --> 19:37.060]  these transformations and actions to work on.
[19:37.500 --> 19:40.860]  And so what happens is if I write this code,
[19:40.860 --> 19:43.700]  thinking that I just multiplied every element by 20,
[19:43.700 --> 19:45.840]  actually nothing gets sent to the workers just yet.
[19:45.840 --> 19:46.700]  Nothing happened.
[19:46.700 --> 19:47.720]  Nothing gets computed.
[19:47.720 --> 19:50.340]  I will have the same results as I had earlier.
[19:50.340 --> 19:51.900]  In fact, I would have nothing.
[19:53.520 --> 19:54.440]  Why?
[19:54.440 --> 19:59.140]  Because when the driver starts parsing this application,
[19:59.140 --> 20:03.460]  all it does is actually it builds an execution graph.
[20:04.880 --> 20:06.060]  And this execution graph is called
[20:06.060 --> 20:07.640]  the DAG and Spark terminology.
[20:07.880 --> 20:09.340]  But basically all it does,
[20:09.340 --> 20:12.420]  it follows all these calls that we made.
[20:12.420 --> 20:15.920]  So, oh yeah, it knows that it's going to call the paralyzed.
[20:15.920 --> 20:18.200]  So it knows that it's going to build an RDD.
[20:18.200 --> 20:19.260]  It doesn't care what's inside.
[20:19.260 --> 20:20.820]  It just knows that it's going to build an RDD
[20:20.820 --> 20:22.780]  with X number of partitions.
[20:22.860 --> 20:24.680]  Then, oh yeah, we're going to call a map.
[20:24.680 --> 20:26.120]  It doesn't care what's inside the map.
[20:26.120 --> 20:27.340]  It doesn't care that we're going to multiply
[20:27.340 --> 20:28.460]  every element by 20.
[20:28.460 --> 20:29.780]  That's not the driver's job.
[20:29.780 --> 20:32.460]  The driver's job is only to follow,
[20:32.460 --> 20:34.000]  to build this execution graph.
[20:34.000 --> 20:35.720]  So it knows that it's going to apply a map.
[20:35.720 --> 20:37.720]  So given a set number of partition,
[20:37.720 --> 20:39.620]  it's going to have an output,
[20:39.620 --> 20:40.680]  the same number of partition.
[20:40.680 --> 20:42.920]  It's going to continue building up this DAG.
[20:42.920 --> 20:44.160]  Maybe it's going to be a filter map,
[20:44.160 --> 20:46.460]  which is another type of transformation, et cetera.
[20:46.500 --> 20:47.940]  And it's going to continue building this graph
[20:47.940 --> 20:49.280]  until it hits an action.
[20:49.280 --> 20:51.640]  An action in Spark is something that's going to force it
[20:51.640 --> 20:53.680]  to do the actual computation.
[20:53.680 --> 20:57.220]  So think like save file or collect or take sample.
[20:57.220 --> 21:00.080]  All these methods that you can call in an RDD
[21:00.080 --> 21:02.900]  that will force it to actually have the right value.
[21:02.960 --> 21:03.900]  And once you do that,
[21:03.900 --> 21:06.840]  once you call a collect or an action in Spark,
[21:07.220 --> 21:08.420]  then, and only then,
[21:08.420 --> 21:10.880]  will the graph be sent to the executors.
[21:11.220 --> 21:14.600]  The graph, mind you, not the code, simply the graph.
[21:14.980 --> 21:16.200]  And it goes a little bit like this.
[21:16.200 --> 21:17.940]  So the driver will send to the workers
[21:18.460 --> 21:19.640]  the graph of execution.
[21:19.640 --> 21:22.400]  The workers will go through this graph
[21:22.400 --> 21:23.720]  and they go, oh, I need to parallelize,
[21:23.720 --> 21:24.680]  but I don't have that list.
[21:24.680 --> 21:25.520]  Oh, you know what?
[21:25.540 --> 21:26.360]  Give me that list.
[21:26.360 --> 21:29.020]  And the driver will send all these serialized objects,
[21:29.020 --> 21:30.220]  one after the other.
[21:30.380 --> 21:31.500]  I need to do a map.
[21:31.500 --> 21:33.500]  Oh, but I don't have the function multiply.
[21:33.740 --> 21:35.320]  You know, they ask the driver to give,
[21:35.320 --> 21:37.780]  serialize the function, the method,
[21:37.780 --> 21:39.220]  and send it to the workers, et cetera.
[21:39.300 --> 21:41.140]  And then they're going to apply those methods.
[21:41.140 --> 21:44.380]  So each worker will only apply these methods,
[21:44.380 --> 21:46.920]  will only follow the DAG on its own partition.
[21:47.500 --> 21:49.360]  So the worker number one will, you know,
[21:49.360 --> 21:51.020]  loop through the elements of the first partition.
[21:51.020 --> 21:54.180]  So one, two, three, multiply by 20, 20, 40, 60,
[21:54.180 --> 21:55.780]  send back the result to worker number two,
[21:55.780 --> 21:56.780]  will do the same.
[21:56.780 --> 21:58.460]  And the driver will aggregate the results
[21:58.460 --> 22:01.160]  and present it to the operator, us.
[22:01.620 --> 22:03.840]  So when you think about it this way,
[22:03.840 --> 22:06.840]  and you go back to our code earlier with our subprocessor,
[22:06.840 --> 22:09.960]  we're just sitting naked there alone in the wilderness.
[22:10.360 --> 22:12.220]  Well, of course it didn't make it to the workers.
[22:12.220 --> 22:14.640]  The workers are only ever aware of the DAG.
[22:14.640 --> 22:17.000]  And so we need, if we want to execute code,
[22:17.000 --> 22:19.320]  we need to somehow embed it inside the DAG.
[22:19.900 --> 22:23.360]  So we need to put it inside a transformation,
[22:23.360 --> 22:24.540]  hence the following code.
[22:24.540 --> 22:26.800]  So here we can see, we define a Lambda function,
[22:26.800 --> 22:27.820]  anonymous function,
[22:27.820 --> 22:32.120]  and inside we put our P open command execution stuff.
[22:32.320 --> 22:34.820]  And obviously we need to follow it by an action,
[22:34.820 --> 22:36.400]  otherwise nothing gets sent.
[22:36.400 --> 22:38.820]  And this is the skeleton that you need to follow,
[22:38.820 --> 22:40.640]  or the basic skeleton you need to follow
[22:40.640 --> 22:43.380]  in order to achieve code execution on Spark
[22:43.380 --> 22:45.620]  across multiple machines.
[22:45.760 --> 22:49.200]  Now, if you don't want to be bothered by creating like,
[22:49.200 --> 22:51.520]  you know, RDDs, handling parallelism and whatnot,
[22:51.520 --> 22:54.460]  I took this skeleton, put it in a fancier way,
[22:54.460 --> 22:55.300]  or something like that.
[22:55.300 --> 22:57.540]  And I incorporated in a tool called Sparky
[22:57.540 --> 22:59.780]  that will take care of all this stuff for you.
[22:59.780 --> 23:01.840]  So you just, you know, point to a cluster,
[23:01.840 --> 23:04.480]  point to the IP that you want to target,
[23:05.100 --> 23:06.460]  send the command that you want,
[23:06.460 --> 23:07.780]  and then how many times,
[23:07.780 --> 23:10.160]  on how many workers you want it to execute,
[23:10.160 --> 23:11.620]  and it will do that for you.
[23:11.620 --> 23:14.120]  It will build the RDDs, build the APIs,
[23:14.120 --> 23:15.080]  call the transformations,
[23:15.080 --> 23:17.020]  the right ones in order to achieve the code,
[23:17.020 --> 23:18.620]  the execution that you want.
[23:19.040 --> 23:20.780]  And this is what it looks like really.
[23:20.780 --> 23:25.400]  So again, I'm going to publish it on GitHub for everyone.
[23:25.460 --> 23:27.500]  So yeah, you'll have a chance to look at the code.
[23:27.500 --> 23:29.020]  So you can see here, we just point to the cluster,
[23:29.020 --> 23:31.060]  we specify our public IP address,
[23:31.060 --> 23:32.560]  the command that we want to execute,
[23:32.560 --> 23:34.860]  and the number of workers that we want to target.
[23:37.000 --> 23:38.520]  And you can see here that we have execution
[23:38.520 --> 23:40.740]  on three machines, so great.
[23:40.980 --> 23:42.360]  I'd also add some, you know,
[23:42.360 --> 23:44.700]  I embedded some scripts to make it easy
[23:44.700 --> 23:46.880]  to run in an AWS environment,
[23:46.880 --> 23:51.040]  to like dump some AWS secrets if you want,
[23:51.040 --> 23:55.760]  or search for files that are sometimes dumped by Spark,
[23:55.760 --> 23:58.200]  you know, work in some work folder.
[23:58.200 --> 24:01.800]  So you can search for AWS keys or secrets or whatever.
[24:01.800 --> 24:05.960]  So it automates a lot of the pen tests in Spark.
[24:05.960 --> 24:11.080]  So yeah, that will get you started.
[24:11.620 --> 24:15.760]  Next, now in this type of execution that I just showed,
[24:15.760 --> 24:18.260]  there are some problems that are quite annoying.
[24:18.260 --> 24:19.640]  Let me go through them very quickly.
[24:19.640 --> 24:22.380]  First of all, the Python version on the driver
[24:22.380 --> 24:24.920]  needs to match the one on the worker,
[24:24.920 --> 24:26.620]  down to the minor version,
[24:26.620 --> 24:28.300]  otherwise it will throw an error.
[24:29.360 --> 24:30.380]  Now there's a way...
[24:30.380 --> 24:32.140]  Yeah, so you can see even if you have a worker
[24:32.140 --> 24:34.860]  that's on 3.5 and your driver is on 3.7,
[24:34.860 --> 24:35.580]  you're kind of screwed.
[24:35.580 --> 24:37.400]  So there's a way around it.
[24:37.740 --> 24:40.420]  Since the check is simply done by checking the version info,
[24:40.420 --> 24:41.840]  you can just override it.
[24:41.840 --> 24:44.360]  So you can see here that I put it like 3.5.
[24:44.920 --> 24:46.660]  And if you do that, it will work,
[24:46.660 --> 24:49.000]  so long as it is not too much of a gap
[24:49.000 --> 24:51.500]  between the two versions.
[24:51.500 --> 24:53.840]  So it can get away with 3.6 and 3.7
[24:53.840 --> 24:57.460]  because the serialized objects are similar to some extent,
[24:57.460 --> 24:59.780]  but it cannot get away with the 3.4 and the 3.7
[24:59.780 --> 25:02.220]  or there's too much of a gap.
[25:02.740 --> 25:05.660]  And the way around that is simply,
[25:05.660 --> 25:07.140]  well, to override the file,
[25:07.140 --> 25:08.760]  but also to take advantage of the fact
[25:08.760 --> 25:11.380]  that a Spark cluster will use, or PySpark,
[25:11.380 --> 25:14.900]  will use pickle deserializers to deserialize the object.
[25:15.280 --> 25:18.160]  And in pickle, you can specify a method.
[25:18.160 --> 25:21.880]  If you define the method reduce of an object,
[25:21.880 --> 25:24.360]  it will get executed first before,
[25:24.360 --> 25:26.540]  in the process of deserialization.
[25:26.540 --> 25:28.940]  So before it fails,
[25:29.370 --> 25:32.280]  that deserialization process later on,
[25:32.280 --> 25:34.660]  it will have executed the command reduce,
[25:34.660 --> 25:36.320]  which contains the execution command.
[25:36.320 --> 25:39.000]  And so you simply need to define the object
[25:39.000 --> 25:40.800]  and send the method as part of that object,
[25:40.800 --> 25:41.760]  and it will get you through.
[25:41.760 --> 25:44.640]  So even though the job submission fails,
[25:44.640 --> 25:45.940]  you still achieve code execution
[25:46.600 --> 25:49.340]  over however many workers you decided.
[25:49.540 --> 25:52.280]  And I did not incorporate it inside Sparky
[25:52.280 --> 25:54.060]  because it's just a hacky way.
[25:54.060 --> 25:56.920]  Really the easiest way is simply to align your versions,
[25:56.920 --> 25:58.620]  your version of Python with that of the workers,
[25:58.620 --> 26:02.500]  but oh well, this is just a fun trick to share.
[26:02.640 --> 26:04.400]  But really the most annoying problem
[26:04.400 --> 26:07.280]  is actually a network problem.
[26:07.340 --> 26:09.800]  See what happens when we register an application,
[26:09.800 --> 26:10.920]  we send an application,
[26:10.920 --> 26:12.280]  as we contact the driver,
[26:13.660 --> 26:16.560]  as we, sorry, the driver contacts the cluster manager
[26:16.560 --> 26:19.640]  on RPC port 7077, as we just saw,
[26:19.640 --> 26:21.680]  and we'll send the register application class
[26:21.680 --> 26:25.520]  along with the parameters, the application's name, et cetera.
[26:25.520 --> 26:27.560]  The cluster manager will receive that application.
[26:27.560 --> 26:29.880]  Oh yeah, I need three, four workers, whatever.
[26:30.740 --> 26:32.100]  Gonna fetch the workers,
[26:32.100 --> 26:35.260]  send them to the driver along with some status data.
[26:35.560 --> 26:36.640]  But then just when you would think
[26:36.640 --> 26:39.740]  that it would be the driver's job to contact the workers,
[26:39.740 --> 26:41.860]  actually, it's the workers
[26:41.860 --> 26:44.700]  who initiate the communication towards the driver.
[26:44.700 --> 26:46.820]  So they will be the one contacting the driver
[26:46.820 --> 26:49.500]  on a random RPC port called schedule import
[26:49.500 --> 26:51.940]  to say, hey, we're ready, et cetera.
[26:52.440 --> 26:55.320]  And that's annoying because you don't necessarily
[26:55.320 --> 26:57.640]  have this flow of communication open
[26:58.260 --> 26:59.400]  because the driver is, you know,
[26:59.400 --> 27:01.500]  your computer is usually behind the net or whatever.
[27:01.500 --> 27:02.640]  So you're kind of screwed.
[27:02.640 --> 27:04.420]  Not only that, the workers will initiate
[27:04.420 --> 27:07.060]  another communication on another port altogether
[27:07.060 --> 27:08.720]  called the block manager port
[27:09.070 --> 27:10.860]  in order to receive some data blocks,
[27:10.860 --> 27:12.920]  you know, request some serialized objects,
[27:12.920 --> 27:14.840]  broadcast some variables and whatnot.
[27:14.840 --> 27:17.320]  So that's annoying.
[27:17.320 --> 27:19.160]  So in Sparky, by default,
[27:19.160 --> 27:21.240]  I basically, I default these settings
[27:21.240 --> 27:23.720]  to some common ports in order to, you know,
[27:23.720 --> 27:27.300]  evade or it works in some settings, but not all.
[27:27.300 --> 27:29.440]  But really the solution is to use
[27:29.440 --> 27:31.220]  what we call a cluster mode.
[27:31.220 --> 27:33.200]  And in a cluster mode,
[27:33.200 --> 27:35.260]  you don't run the driver on your laptop.
[27:35.380 --> 27:38.240]  You simply contact the cluster manager on port 77,
[27:38.240 --> 27:39.400]  7077, yeah.
[27:39.400 --> 27:40.500]  And then you tell it,
[27:40.500 --> 27:42.160]  oh, here's the application to execute.
[27:42.160 --> 27:46.660]  And please also set up the driver on your own cluster.
[27:47.860 --> 27:49.540]  And that works pretty well.
[27:49.540 --> 27:52.740]  So in this case, we only need access to port 7077.
[27:52.940 --> 27:56.320]  The downside is that you cannot use PySpark.
[27:56.320 --> 27:57.580]  You cannot use Python.
[27:57.580 --> 28:01.220]  So I had to redo that skeleton I showed you earlier
[28:01.220 --> 28:01.860]  in Python.
[28:01.860 --> 28:05.520]  I had to do it in Scala to build a jar file
[28:05.520 --> 28:07.580]  that will parallelize,
[28:07.580 --> 28:10.240]  that will execute whatever command you give it
[28:10.240 --> 28:13.620]  across however many servers that you want.
[28:13.620 --> 28:16.080]  So yeah, the code is again on GitHub,
[28:16.080 --> 28:18.500]  but you don't need to touch this one.
[28:18.500 --> 28:20.400]  It's this compiled jar file.
[28:21.240 --> 28:23.580]  So you can just use it as it is.
[28:23.760 --> 28:25.200]  It's, yeah.
[28:25.460 --> 28:27.480]  So here's a quick example.
[28:27.480 --> 28:28.900]  So you can see we host this jar file
[28:28.900 --> 28:30.560]  that I just showed you on a server
[28:30.560 --> 28:32.460]  that is reachable by the workers.
[28:32.460 --> 28:35.760]  And we simply use Sparky again to build that command
[28:35.760 --> 28:37.420]  to register this in cluster mode.
[28:37.420 --> 28:39.160]  So we just point to this cluster,
[28:39.160 --> 28:41.560]  to this jar file command we want to execute.
[28:41.560 --> 28:44.680]  And then we don't need any network condition
[28:44.680 --> 28:47.960]  except for that 7077 to reach the cluster.
[28:47.960 --> 28:49.660]  And it works pretty well.
[28:49.840 --> 28:51.660]  It will even auto delete itself.
[28:51.660 --> 28:53.180]  So you don't have to worry about that.
[28:53.180 --> 28:56.320]  So this gets us through the network thing.
[28:56.780 --> 28:58.620]  So these are the main ways to execute code
[28:58.620 --> 29:02.520]  on Spark that I found on a Spark cluster
[29:02.520 --> 29:06.280]  that has no authentication, which is enabled by default.
[29:06.420 --> 29:10.040]  So anyway, let's go explore some other facets of Spark.
[29:10.260 --> 29:13.140]  And there's this interesting interface called,
[29:13.140 --> 29:14.660]  well, it's the REST API.
[29:14.660 --> 29:15.640]  And it is interesting
[29:15.640 --> 29:17.980]  because when you go through the documentation,
[29:17.980 --> 29:20.840]  you can see that it's available in port 6066.
[29:21.160 --> 29:22.660]  And when you look at the documentation,
[29:22.660 --> 29:25.960]  it's really a read-only boring API.
[29:25.960 --> 29:27.480]  I was like, yeah, whatever.
[29:27.480 --> 29:29.400]  And I remember a few days later,
[29:29.400 --> 29:30.760]  I was digging through the source code,
[29:30.760 --> 29:33.180]  like really lost inside the source code
[29:33.180 --> 29:37.280]  amongst weird scholarship functions, options and whatnot.
[29:37.280 --> 29:40.080]  And I remember I saw this keyword REST.
[29:40.080 --> 29:41.660]  I'm like, oh my God, I know what that is.
[29:41.660 --> 29:43.460]  And I grabbed it and I followed it.
[29:43.480 --> 29:45.640]  And I came across this piece of code
[29:45.640 --> 29:48.920]  that says something like, submit requests, create.
[29:48.920 --> 29:50.600]  I was like, I like to submit stuff.
[29:50.600 --> 29:51.760]  I like to create stuff.
[29:51.760 --> 29:52.900]  So this thing do.
[29:53.980 --> 29:57.100]  And you follow this, create submission requests.
[29:57.100 --> 30:00.000]  And I was like, let's talk about app resource, main class.
[30:00.000 --> 30:01.300]  I'm like, well, hang on.
[30:01.300 --> 30:05.840]  App resources references a jar usually in Spark code.
[30:05.840 --> 30:08.500]  So I thought this was a read-only API.
[30:08.500 --> 30:09.620]  What the hell is going on?
[30:09.620 --> 30:12.560]  So I followed this, create submission request.
[30:12.560 --> 30:13.580]  And I came across this.
[30:13.580 --> 30:16.080]  Yeah, app resources, indeed an application jar.
[30:16.080 --> 30:17.720]  So basically the REST API,
[30:17.720 --> 30:18.860]  even though the documentation,
[30:18.860 --> 30:20.700]  it mentions it's only a read-only API,
[30:20.700 --> 30:22.160]  actually in the code,
[30:22.160 --> 30:24.380]  it accepts a jar file that you can send it,
[30:25.320 --> 30:26.560]  which is kind of amazing.
[30:27.040 --> 30:30.440]  All you have to do is simply build a JSON file.
[30:30.440 --> 30:31.820]  It accepts a JSON input.
[30:31.820 --> 30:34.000]  So you build a JSON file with the right properties.
[30:34.000 --> 30:37.480]  So the app resource, main class, Spark properties and so on.
[30:37.480 --> 30:39.660]  And then you can send your jar file
[30:39.660 --> 30:40.940]  and then chief code execution.
[30:40.940 --> 30:44.240]  And we can take that jar file we used earlier.
[30:44.240 --> 30:46.520]  So simple app jar, send it a command arc.
[30:46.520 --> 30:49.840]  So this is, it just accepts commands to execute in base 64.
[30:49.840 --> 30:52.000]  So this is really, I think a, who am I,
[30:52.000 --> 30:53.520]  or some shit like, no, not who am I,
[30:53.520 --> 30:57.760]  but touch some file in some folder in base 64.
[30:57.760 --> 30:58.580]  It's very simple.
[30:59.080 --> 31:00.580]  And yeah, but there's different properties
[31:00.580 --> 31:02.760]  and you can just curl this payload
[31:02.760 --> 31:05.620]  and it will execute codes as simple as that.
[31:06.480 --> 31:09.000]  And this is just to show you what it looks like.
[31:09.560 --> 31:13.380]  But yeah, but again, if you use this method,
[31:13.380 --> 31:15.520]  the driver, you don't need it to run on your machine.
[31:15.520 --> 31:17.040]  It's going to run in the cluster.
[31:17.040 --> 31:18.320]  So it's going to run in cluster mode.
[31:18.320 --> 31:19.660]  And one thing I will mention,
[31:19.660 --> 31:22.800]  if you decide to use the REST API or cluster mode in general,
[31:22.800 --> 31:24.440]  you can send whatever jar file you want.
[31:24.440 --> 31:26.600]  You don't have to use my specific jar file,
[31:26.600 --> 31:29.200]  but if you, you will get execution on one worker.
[31:29.200 --> 31:31.180]  But if you want to execute on 200, 300,
[31:31.180 --> 31:33.160]  all the workers inside the cluster,
[31:33.160 --> 31:34.760]  then you need to use Spark API.
[31:34.760 --> 31:39.320]  So you need to use some version of the Spark,
[31:39.320 --> 31:41.300]  like, you know, Spark APIs I just showed
[31:41.300 --> 31:44.840]  in the Scala code earlier, or Java code earlier.
[31:44.840 --> 31:48.160]  So anyway, so this was fun and I thought,
[31:48.160 --> 31:50.220]  well, I wonder if anybody blogged about it.
[31:50.220 --> 31:52.780]  And I Googled create submission request.
[31:52.780 --> 31:55.420]  And I came across the first result on Google
[31:55.420 --> 31:57.580]  was this script and it was already published.
[31:57.580 --> 31:59.640]  And I was like, fuck, so close.
[31:59.640 --> 32:01.660]  Not only that, there was actually an exploit
[32:01.660 --> 32:04.280]  that was already, you know, published.
[32:04.280 --> 32:05.900]  It was actually a Metasploit model
[32:06.440 --> 32:08.660]  that was published by these fine gentlemen.
[32:08.660 --> 32:09.920]  I was like, damn it.
[32:10.180 --> 32:13.240]  Just, you know, sometimes you keep your head in the wheel
[32:13.240 --> 32:15.200]  and you forget to do basic research,
[32:15.200 --> 32:17.800]  but it was just, you know, months away.
[32:17.960 --> 32:21.460]  Anyway, let's get to the real stuff, right?
[32:22.840 --> 32:24.040]  So authentication.
[32:24.280 --> 32:26.280]  Let's say you come across a cluster
[32:26.280 --> 32:29.180]  that has all the right switches on, right?
[32:29.180 --> 32:30.800]  So it's enabled authentication.
[32:30.800 --> 32:32.880]  And the way to do that in Spark is fairly easy.
[32:32.880 --> 32:35.200]  You just enable authenticate to true.
[32:35.200 --> 32:37.500]  Then you define a secret, you can see here.
[32:38.040 --> 32:40.120]  And yeah, all the components you do,
[32:40.120 --> 32:42.100]  you need to do that on every component inside the cluster.
[32:42.100 --> 32:46.320]  So every worker and every cluster manager and the driver.
[32:46.360 --> 32:48.000]  And that way they all have a shared secret
[32:48.000 --> 32:48.820]  to communicate with.
[32:48.820 --> 32:50.500]  And if you have an authenticated cluster
[32:50.500 --> 32:52.720]  and you try to communicate with it,
[32:52.720 --> 32:55.400]  you will get this delightful error that says,
[32:55.400 --> 32:57.980]  you know, you send this check existence class,
[32:57.980 --> 33:00.820]  and then you get this illegal state exception, ACSL message.
[33:00.820 --> 33:03.140]  Now, ACSL is an authentication protocol,
[33:03.140 --> 33:04.460]  challenge response protocol,
[33:05.480 --> 33:09.320]  based on a shared secret, as you might've guessed.
[33:09.920 --> 33:12.460]  But yeah, basically the server sends a nonce to the client
[33:12.460 --> 33:14.700]  and the client hashes that with the nonce,
[33:14.700 --> 33:15.860]  hashes the secrets with the nonce,
[33:15.860 --> 33:17.840]  and then sends the response plus other parameters
[33:17.840 --> 33:20.240]  that we will see, but fairly simple.
[33:20.240 --> 33:22.500]  But the funny thing is that even though the driver,
[33:22.500 --> 33:24.420]  you configure the driver to use the secret
[33:24.420 --> 33:26.860]  and you tell it to authenticate,
[33:26.860 --> 33:30.020]  it will first attempt an unauthenticated session.
[33:30.020 --> 33:34.480]  So it will first, you know, try check existence,
[33:34.480 --> 33:36.080]  get slapped with this error,
[33:36.080 --> 33:38.260]  and then it will try an authenticated message.
[33:38.260 --> 33:39.020]  So it's kind of weird.
[33:39.020 --> 33:40.760]  So anyway, what does it look like,
[33:40.760 --> 33:42.080]  an authenticated session?
[33:42.080 --> 33:45.540]  Well, first the driver will send 21 bytes of data,
[33:45.540 --> 33:46.900]  the header, the usual header,
[33:46.900 --> 33:50.460]  with the magic bytes, this time set to 2B03,
[33:50.460 --> 33:52.660]  as opposed to C05 earlier,
[33:52.660 --> 33:54.700]  followed by, you know, eight random bytes
[33:54.700 --> 33:56.480]  that we don't care about, the size of the payload,
[33:56.480 --> 33:58.060]  which is definitely wrong in this case,
[33:58.060 --> 34:02.140]  because we're not talking, it's not 176 bytes.
[34:02.140 --> 34:03.500]  But anyway, the most important thing
[34:03.500 --> 34:04.720]  in this authentication sequence
[34:04.720 --> 34:07.520]  is that it sends the Spark ACSL user,
[34:07.520 --> 34:08.480]  which is going to be the user
[34:08.480 --> 34:12.480]  used inside the challenge calculation.
[34:12.920 --> 34:15.340]  So anyway, Cluster Manager responds with 21 bytes data
[34:15.340 --> 34:16.960]  as usual, the header,
[34:16.960 --> 34:19.740]  and then it follows by the parameters to,
[34:19.740 --> 34:22.720]  ACSL parameters to perform the challenge computation.
[34:22.720 --> 34:26.560]  So you have the nonce, random value, realm, QOP,
[34:26.560 --> 34:27.980]  which is the, are we doing authentication
[34:27.980 --> 34:29.680]  or encryption or both?
[34:29.680 --> 34:31.740]  These delightful algorithms to use
[34:31.740 --> 34:34.180]  in case we're doing ACSL encryption.
[34:34.180 --> 34:35.900]  Now, just to reassure you,
[34:35.900 --> 34:39.800]  Spark does not use ACSL encryption by default.
[34:39.980 --> 34:41.780]  It uses AES, but it's definitely possible
[34:41.780 --> 34:43.820]  to use it as a fallback.
[34:43.820 --> 34:46.480]  And the algorithm, and this one is indeed used
[34:46.980 --> 34:48.420]  to calculate the hash.
[34:48.420 --> 34:49.620]  And the calculation of the hash,
[34:49.620 --> 34:51.040]  we're not going to do it, honestly.
[34:52.100 --> 34:54.160]  It's detailed in RFC 2831,
[34:54.160 --> 34:55.700]  but you take a bunch of data,
[34:55.700 --> 34:57.620]  you hash it in MD5, you take the output,
[34:57.620 --> 34:59.380]  you combine it with the nonce and other stuff.
[34:59.380 --> 35:01.380]  And then you hashed a bunch,
[35:01.380 --> 35:03.080]  you hash other types of data,
[35:03.080 --> 35:05.400]  and then you combine everything,
[35:05.400 --> 35:07.720]  which gives you the response to the challenge.
[35:07.760 --> 35:09.820]  Again, if you want to know more about it,
[35:09.820 --> 35:11.320]  Google it, RFC 2831.
[35:11.320 --> 35:14.880]  So, the driver sends the response
[35:14.880 --> 35:16.260]  to the cluster manager.
[35:16.840 --> 35:19.500]  If everything checks out, the cluster manager acts.
[35:20.100 --> 35:22.320]  Right, and then it can follow
[35:23.260 --> 35:26.320]  the classic flow of communication.
[35:26.320 --> 35:28.540]  So basically, check existence class,
[35:28.540 --> 35:31.240]  register application, register with application,
[35:32.940 --> 35:35.140]  executors added, yada, yada, yada.
[35:35.560 --> 35:36.880]  Now, here's what bothered me,
[35:36.880 --> 35:39.160]  or what bothered me, what ticked me a little bit,
[35:39.160 --> 35:43.380]  is that this authentication step was only ever done
[35:43.380 --> 35:47.320]  when there was an RPC endpoint that was prefixed,
[35:47.320 --> 35:49.660]  an RPC-ish communication that was prefixed
[35:49.660 --> 35:51.640]  with those 21 bytes of data.
[35:51.640 --> 35:54.420]  But it turns out, there's another header
[35:54.420 --> 35:57.280]  that is sometimes sent before payloads,
[35:57.280 --> 35:59.960]  that prefixes payloads.
[35:59.960 --> 36:02.520]  And this header is only 13 bytes long.
[36:02.520 --> 36:05.580]  And it's usually sent inside the middle,
[36:05.580 --> 36:07.500]  inside the TCP stream.
[36:07.500 --> 36:09.400]  So it's inside the same TCP session.
[36:09.400 --> 36:11.220]  So I did not have much hope.
[36:11.220 --> 36:13.420]  I was like, well, maybe the cluster actually remembers
[36:13.420 --> 36:15.260]  that it's in an authenticated session.
[36:15.260 --> 36:17.120]  So that's why, et cetera.
[36:17.140 --> 36:19.580]  But nevertheless, I thought, well, you know what?
[36:19.580 --> 36:21.140]  Let's see what it looks like.
[36:21.140 --> 36:24.440]  Let's see if we start the communication from here.
[36:24.440 --> 36:27.100]  So we don't send this check existence header,
[36:27.100 --> 36:29.440]  which prompts the authentication message.
[36:29.440 --> 36:32.120]  Let's just start communication from here,
[36:32.120 --> 36:33.680]  by sending these 13 bytes of data,
[36:33.680 --> 36:35.720]  followed by the register application,
[36:35.720 --> 36:37.320]  because that's what we really care about, right?
[36:37.320 --> 36:38.960]  We want to submit an application.
[36:39.560 --> 36:40.880]  That's what we really want.
[36:41.180 --> 36:44.820]  And so if you begin communication at this stage,
[36:44.820 --> 36:47.080]  you first send 13 bytes of data.
[36:47.580 --> 36:51.320]  It looks awfully similar to the 21 bytes header,
[36:51.320 --> 36:52.720]  but basically, oh, sorry.
[36:52.840 --> 36:55.300]  You have four bytes, four null bytes,
[36:55.300 --> 36:57.540]  followed by the size of the payload plus the header.
[36:57.540 --> 37:00.380]  So the size plus 13, a magic byte nine,
[37:00.380 --> 37:01.900]  which didn't change for some reason,
[37:01.900 --> 37:03.500]  and four bytes of data,
[37:03.500 --> 37:04.900]  which are the actual size of the payload.
[37:04.900 --> 37:07.340]  And then you send the serialized register application
[37:07.340 --> 37:10.000]  with all its attributes, you know, application's name,
[37:10.000 --> 37:10.420]  yeah, yeah.
[37:10.900 --> 37:14.060]  And I sent this, and I was expecting a massive error,
[37:14.060 --> 37:16.400]  like, you know, those JVM hundred line errors,
[37:16.400 --> 37:19.540]  but instead I got a header
[37:19.540 --> 37:21.800]  followed by register application class.
[37:21.800 --> 37:23.760]  And I was like, is it me,
[37:23.760 --> 37:25.920]  or is it just possible to bypass authentication?
[37:25.920 --> 37:28.200]  I was like, hallelujah, it's amazing.
[37:28.940 --> 37:31.640]  Now, bypassing authentication is fine,
[37:31.640 --> 37:33.860]  but what can we do with it?
[37:34.420 --> 37:37.660]  Because I was not going to recreate the 80 next messages
[37:37.660 --> 37:41.300]  that were following along, it's not possible.
[37:41.520 --> 37:43.980]  So what can we do with just one message?
[37:43.980 --> 37:47.580]  Well, if you look back at Spark, how it works,
[37:47.580 --> 37:49.500]  the register application class is the one
[37:49.500 --> 37:54.120]  that prompts all that workload on the cluster manager
[37:54.120 --> 37:56.700]  that's gonna contact the workers and say,
[37:56.700 --> 37:58.540]  hey, I've got an application for you,
[37:58.540 --> 38:00.940]  you better get ready, spawn those executors,
[38:00.940 --> 38:02.700]  and get ready to take the work.
[38:03.300 --> 38:06.240]  So I went ahead to the workers and I replayed my message,
[38:06.240 --> 38:08.520]  hoping to see some process spawn or whatever,
[38:08.520 --> 38:10.120]  but I didn't see anything.
[38:10.540 --> 38:13.600]  And the reason for it, I figured out a couple of days later,
[38:13.600 --> 38:16.440]  was because when you send this payload,
[38:16.440 --> 38:19.260]  serialized register application, when you send it,
[38:19.260 --> 38:21.200]  the cluster manager receives it.
[38:22.320 --> 38:24.660]  I immediately shut down the connection,
[38:24.660 --> 38:25.620]  the program terminated.
[38:25.620 --> 38:27.860]  So the cluster manager did not have enough time
[38:27.860 --> 38:29.880]  to actually contact the worker,
[38:29.880 --> 38:32.900]  spawn the process, and let it do its thing.
[38:33.160 --> 38:36.260]  Now, so I immediately shut down the connection.
[38:36.260 --> 38:38.660]  So the way around it, in order to give time,
[38:38.660 --> 38:40.220]  you simply need to do a sleep.
[38:40.220 --> 38:41.420]  As simple as that.
[38:41.420 --> 38:42.900]  You send the register application,
[38:42.900 --> 38:44.400]  you sleep for five seconds.
[38:44.420 --> 38:46.860]  That way you give enough time for the cluster manager
[38:46.860 --> 38:50.540]  to contact the worker, spawn the executor, and do its work.
[38:50.540 --> 38:52.600]  And once you do that, once you do that sleep,
[38:52.600 --> 38:54.640]  I saw, like I put a watcher on the worker,
[38:54.640 --> 38:56.220]  monitoring for processes,
[38:56.220 --> 39:00.160]  and I saw this process getting started.
[39:00.280 --> 39:02.400]  And the magnificent thing about it was that
[39:02.400 --> 39:06.180]  it took as parameters, something that I was,
[39:06.180 --> 39:09.700]  I was defining inside that serialized register application.
[39:09.700 --> 39:11.920]  You know, the block manager report set to 8443.
[39:11.920 --> 39:14.260]  And I knew it was me because I'm the one that said it
[39:14.260 --> 39:16.660]  in order to have communication come from the worker
[39:16.660 --> 39:18.220]  to the driver.
[39:18.580 --> 39:22.440]  So I thought, okay, authentication bypass is a sure thing.
[39:22.860 --> 39:24.960]  Remote code execution is a definite maybe.
[39:24.960 --> 39:26.360]  Because you can influence parameters
[39:26.360 --> 39:28.560]  that are being used to spawn a process.
[39:28.560 --> 39:30.380]  So maybe there's some injection there.
[39:30.380 --> 39:33.020]  So I turned to this register application class
[39:33.020 --> 39:35.020]  and saw like, okay, what's inside?
[39:35.020 --> 39:37.200]  What can we define as parameters as options
[39:37.200 --> 39:38.700]  that we can take an advantage of?
[39:38.700 --> 39:40.800]  So register application, again, a case class,
[39:40.800 --> 39:42.100]  which extends the deploy message,
[39:42.100 --> 39:44.360]  which implements the trade serializable.
[39:44.460 --> 39:45.720]  Don't care about that.
[39:45.720 --> 39:47.560]  Application description, what does it do?
[39:47.560 --> 39:49.440]  Okay, so if you're like me,
[39:49.440 --> 39:50.920]  you're going to ignore all this shit
[39:50.920 --> 39:53.960]  and go straight to command, right?
[39:53.960 --> 39:56.840]  Because the command class, so command case class,
[39:56.840 --> 39:57.880]  it has the main class,
[39:57.880 --> 40:00.060]  which is the class of the executor to be executed.
[40:00.060 --> 40:01.580]  We cannot overwrite that.
[40:01.620 --> 40:04.280]  You have arguments, environments, a bunch of parameters.
[40:04.320 --> 40:06.860]  Now, if you just want to inject like, you know,
[40:07.060 --> 40:09.320]  a semicolon or an and or a pipe,
[40:09.320 --> 40:13.180]  it will not work because these parameters are sanitized.
[40:13.600 --> 40:15.980]  So we need a clever, more clever way
[40:15.980 --> 40:17.700]  of hijacking the execution.
[40:17.700 --> 40:18.660]  And the way to do it,
[40:18.660 --> 40:20.580]  I found the easiest way to do it at least,
[40:20.580 --> 40:22.540]  is using these Java options.
[40:22.540 --> 40:24.180]  Now, if you're familiar with the JVM,
[40:24.180 --> 40:26.100]  you know that you can actually specify
[40:26.100 --> 40:28.300]  or control the behavior of that JVM.
[40:28.300 --> 40:30.540]  Like, I don't know which garbage collector to use,
[40:30.540 --> 40:33.620]  how should strings be compressed,
[40:33.620 --> 40:35.720]  the amount of heap memory, et cetera.
[40:35.920 --> 40:39.120]  Well, it turns out if you dig into the hundreds
[40:39.120 --> 40:40.380]  and hundreds of Java options,
[40:40.380 --> 40:43.180]  there is one actually that allow code execution.
[40:43.280 --> 40:45.660]  Also, yeah, this is an example of a JVM option.
[40:45.840 --> 40:47.660]  So there is one that allows code execution
[40:47.660 --> 40:50.100]  and it's called an un-out-of-memory error.
[40:50.100 --> 40:52.760]  And you just specify a command and it will execute it.
[40:52.760 --> 40:54.940]  The catch is that it only executes it
[40:54.940 --> 40:56.960]  if there's an un-out-of-memory error.
[40:56.960 --> 40:58.800]  So like its name said.
[40:59.120 --> 41:04.680]  So if the executor or process fails to allocate memory,
[41:04.680 --> 41:06.500]  heap memory, then it will throw this error
[41:06.500 --> 41:07.660]  and then it will execute code.
[41:07.660 --> 41:10.520]  So how to make the executor trigger this error?
[41:10.520 --> 41:14.260]  Well, simple way is simply to add another JVM option, xmx,
[41:14.260 --> 41:16.500]  which will set the maximum amount of memory
[41:16.500 --> 41:19.300]  to a ridiculous amount, like one megabyte or two megabytes.
[41:19.300 --> 41:22.080]  And once you combine these two properties,
[41:22.080 --> 41:23.920]  then you have your code execution.
[41:24.580 --> 41:26.720]  And so ideally, we'd like to do something like this.
[41:26.720 --> 41:28.480]  Obviously, we cannot do it inside the code
[41:28.480 --> 41:31.480]  because while the driver will attempt to authenticate,
[41:31.480 --> 41:32.340]  which is no good.
[41:32.340 --> 41:34.940]  And anyway, you cannot set the xmx inside the driver.
[41:34.940 --> 41:38.180]  So we really need to forge by hand, basically,
[41:38.180 --> 41:40.900]  that serialized object, embed these options,
[41:40.900 --> 41:42.300]  have a properly serialized object
[41:42.300 --> 41:45.940]  and send it to the cluster manager.
[41:45.940 --> 41:49.820]  And again, easy way to do it using Sparky.
[41:51.780 --> 41:53.300]  So I just point...
[41:53.300 --> 41:54.660]  So this is just a quick example
[41:54.660 --> 41:56.420]  to show you how to exploit this phone.
[41:57.220 --> 42:00.420]  So this is the reversal that we want to execute, right?
[42:00.420 --> 42:03.100]  So I just point to Sparky, I want to execute this code,
[42:03.100 --> 42:06.060]  but as you can see, it requires,
[42:06.060 --> 42:07.660]  the cluster requires authentication.
[42:09.820 --> 42:12.580]  But if I add dash B, then it will take care,
[42:12.580 --> 42:13.920]  it will take advantage of this phone
[42:13.920 --> 42:15.540]  and you can see it bypassed authentication,
[42:15.540 --> 42:17.320]  which has executed code directly.
[42:17.320 --> 42:21.020]  And it's gonna execute it on as many workers as requested.
[42:21.700 --> 42:23.620]  So yeah, we just bypassed authentication,
[42:23.620 --> 42:25.720]  which has executed code on workers.
[42:26.740 --> 42:27.840]  So great.
[42:28.920 --> 42:30.420]  So yeah, so now that we understood
[42:30.420 --> 42:34.100]  like the hacker intuitive way of finding this phone,
[42:34.100 --> 42:35.480]  let's go through the code,
[42:35.480 --> 42:36.980]  because I think it's interesting.
[42:37.040 --> 42:38.960]  If you see, like, I made the distinction
[42:38.960 --> 42:41.300]  between the 21 bytes, the, you know,
[42:41.960 --> 42:43.960]  payload that was prefixed with 21 bytes of data
[42:43.960 --> 42:45.520]  or prefixed with 13 bytes of data.
[42:45.520 --> 42:49.000]  It turns out there's actually a difference in the dispatcher.
[42:49.040 --> 42:50.680]  So the register application is handled
[42:50.680 --> 42:52.960]  by the process one-way message.
[42:53.020 --> 42:54.360]  And the process one-way message called
[42:54.360 --> 42:56.580]  the receive method of the RPC handler, right?
[42:56.580 --> 42:58.680]  So it delegates to the receive,
[42:58.680 --> 43:00.380]  and you can see there at no point
[43:00.380 --> 43:02.540]  there is any authentication going on here.
[43:02.540 --> 43:04.900]  The delegate receive will actually perform the work.
[43:04.900 --> 43:08.040]  If you compare that with the other receive method,
[43:08.040 --> 43:10.280]  which actually, you know, asks for authentication,
[43:10.280 --> 43:13.680]  you can see that it's quite beefy and different.
[43:13.680 --> 43:15.860]  So the correction, basically,
[43:15.860 --> 43:18.900]  the patch was to protect that same receive method
[43:18.900 --> 43:21.280]  and another one used in streaming,
[43:21.280 --> 43:22.740]  which was affected also,
[43:22.740 --> 43:27.100]  by the same type, the same types of checks.
[43:27.220 --> 43:30.520]  And this was labeled CB20209480.
[43:30.520 --> 43:32.780]  I disclosed them on 24th of December,
[43:32.780 --> 43:36.100]  if I'm not mistaken, which was very horrible for me.
[43:36.100 --> 43:39.640]  But anyway, and yeah, it was fixed by the Apache Spark team.
[43:39.640 --> 43:41.780]  So thank you very much for the work that you guys have done.
[43:41.780 --> 43:42.720]  It's amazing.
[43:43.380 --> 43:48.180]  Yeah, so in summary, what I wanna say is Spark is awesome.
[43:48.280 --> 43:50.440]  I'm gonna say too bad security is not taken seriously,
[43:50.440 --> 43:51.720]  and I'm not talking about the folks
[43:51.720 --> 43:53.180]  who manage security there.
[43:53.480 --> 43:55.020]  I'm just referring to the fact
[43:55.020 --> 43:57.300]  that security is off by default.
[43:57.300 --> 43:58.620]  I don't think that should be something
[43:58.620 --> 44:00.020]  that we should have in 2020,
[44:00.020 --> 44:02.160]  especially for something that's a framework
[44:02.160 --> 44:04.660]  that does, you know, data computation.
[44:04.940 --> 44:06.740]  And we all know how data is valuable.
[44:06.740 --> 44:10.960]  So I say it in reference to that.
[44:11.640 --> 44:13.660]  Anyway, I hope that will change.
[44:13.860 --> 44:16.100]  And finally, yeah, we only covered Spark standalone
[44:16.100 --> 44:18.180]  where the cluster manager is actually
[44:19.400 --> 44:20.820]  the one doing the work.
[44:20.820 --> 44:22.120]  But you can have other setups
[44:22.120 --> 44:24.320]  where it's actually Yarn that's doing the work,
[44:24.320 --> 44:28.040]  or I think even a Kubernetes API server in Spark 3.
[44:28.040 --> 44:30.420]  So yeah, there's a lot of ground to cover.
[44:30.420 --> 44:32.680]  So if you wanna dig into it, go ahead.
[44:33.560 --> 44:34.840]  Yeah, the code to Sparky,
[44:34.840 --> 44:37.480]  if you wanna check out these, you know, Spark APIs,
[44:37.480 --> 44:38.900]  the serialized object that's being forced
[44:38.900 --> 44:41.860]  in order to bypass the vuln, all that stuff, it's there.
[44:41.860 --> 44:46.320]  I hope you got enough of this talk to actually dig into it
[44:46.320 --> 44:47.520]  because that's really the only way
[44:47.520 --> 44:49.500]  to figure out how stuff works.
[44:49.520 --> 44:52.460]  So yeah, your contribution are most welcome.
[44:52.460 --> 44:53.500]  And if you wanna hit me up on Twitter
[44:53.500 --> 44:55.960]  to talk about it more, please do not hesitate.
[44:55.960 --> 44:57.320]  And looking forward to talk to you guys
[44:57.320 --> 44:59.500]  and to talk to you folks in the QA.
[44:59.500 --> 45:00.520]  Thank you very much.
[45:00.520 --> 45:01.200]  Bye-bye.