diff --git a/Gemfile b/Gemfile
index 3b6f3fdaed7..038671ed5ba 100644
--- a/Gemfile
+++ b/Gemfile
@@ -12,6 +12,7 @@ gem 'eventmachine', '~> 1.0.9'
 gem 'google-api-client', '~> 0.8.6' # required for fog-google
 gem 'httpclient'
 gem 'i18n'
+gem 'json-schema'
 gem 'loggregator_emitter', '~> 5.0'
 gem 'membrane', '~> 1.0'
 gem 'mime-types', '~> 2.6.2'
diff --git a/Gemfile.lock b/Gemfile.lock
index 2f2c8dad89c..1966eb73841 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -204,6 +204,8 @@ GEM
     i18n (0.7.0)
     ipaddress (0.8.3)
     json (1.8.3)
+    json-schema (2.8.0)
+      addressable (>= 2.4)
     json_pure (1.8.6)
     jwt (1.5.6)
     launchy (2.4.3)
@@ -437,6 +439,7 @@ DEPENDENCIES
   google-api-client (~> 0.8.6)
   httpclient
   i18n
+  json-schema
   loggregator_emitter (~> 5.0)
   machinist (~> 1.0.6)
   membrane (~> 1.0)
diff --git a/lib/services/service_brokers/v2/catalog_schemas.rb b/lib/services/service_brokers/v2/catalog_schemas.rb
index 8353f237931..374742e08e6 100644
--- a/lib/services/service_brokers/v2/catalog_schemas.rb
+++ b/lib/services/service_brokers/v2/catalog_schemas.rb
@@ -1,10 +1,13 @@
+require 'json-schema'
+
 module VCAP::Services::ServiceBrokers::V2
+  MAX_SCHEMA_SIZE = 65_536
   class CatalogSchemas
     attr_reader :errors, :create_instance
 
-    def initialize(attrs)
+    def initialize(schema)
       @errors = VCAP::Services::ValidationErrors.new
-      validate_and_populate_create_instance(attrs)
+      validate_and_populate_create_instance(schema)
     end
 
     def valid?
@@ -13,26 +16,94 @@ def valid?
 
     private
 
-    def validate_and_populate_create_instance(attrs)
-      return unless attrs
-      unless attrs.is_a? Hash
-        errors.add("Schemas must be a hash, but has value #{attrs.inspect}")
+    def validate_and_populate_create_instance(schemas)
+      return unless schemas
+      unless schemas.is_a? Hash
+        errors.add("Schemas must be a hash, but has value #{schemas.inspect}")
         return
       end
 
       path = []
       ['service_instance', 'create', 'parameters'].each do |key|
         path += [key]
-        attrs = attrs[key]
-        return nil unless attrs
+        schemas = schemas[key]
+        return nil unless schemas
 
-        unless attrs.is_a? Hash
-          errors.add("Schemas #{path.join('.')} must be a hash, but has value #{attrs.inspect}")
+        unless schemas.is_a? Hash
+          errors.add("Schemas #{path.join('.')} must be a hash, but has value #{schemas.inspect}")
           return nil
         end
       end
 
-      @create_instance = attrs
+      create_instance_schema = schemas
+      create_instance_path = path.join('.')
+
+      validate_schema(create_instance_path, create_instance_schema)
+      return unless errors.empty?
+
+      @create_instance = create_instance_schema
+    end
+
+    def validate_schema(path, schema)
+      schema_validations.each do |validation|
+        break if errors.present?
+        send(validation, path, schema)
+      end
+    end
+
+    def schema_validations
+      [
+        :validate_schema_size,
+        :validate_metaschema,
+        :validate_no_external_references,
+        :validate_schema_type
+      ]
+    end
+
+    def validate_schema_type(path, schema)
+      add_schema_error_msg(path, 'must have field "type", with value "object"') if schema['type'] != 'object'
+    end
+
+    def validate_schema_size(path, schema)
+      errors.add("Schema #{path} is larger than 64KB") if schema.to_json.length > MAX_SCHEMA_SIZE
+    end
+
+    def validate_metaschema(path, schema)
+      JSON::Validator.schema_reader = JSON::Schema::Reader.new(accept_uri: false, accept_file: false)
+      file = File.read(JSON::Validator.validator_for_name('draft4').metaschema)
+
+      metaschema = JSON.parse(file)
+
+      begin
+        errors = JSON::Validator.fully_validate(metaschema, schema)
+      rescue => e
+        add_schema_error_msg(path, e)
+        return nil
+      end
+
+      errors.each do |error|
+        add_schema_error_msg(path, "Must conform to JSON Schema Draft 04: #{error}")
+      end
+    end
+
+    def validate_no_external_references(path, schema)
+      JSON::Validator.schema_reader = JSON::Schema::Reader.new(accept_uri: false, accept_file: false)
+
+      begin
+        JSON::Validator.validate!(schema, {})
+      rescue JSON::Schema::SchemaError => e
+        add_schema_error_msg(path, "Custom meta schemas are not supported: #{e}")
+      rescue JSON::Schema::ReadRefused => e
+        add_schema_error_msg(path, "No external references are allowed: #{e}")
+      rescue JSON::Schema::ValidationError
+        # We don't care if our input fails validation on broker schema
+      rescue => e
+        add_schema_error_msg(path, e)
+      end
+    end
+
+    def add_schema_error_msg(path, err)
+      errors.add("Schema #{path} is not valid. #{err}")
     end
   end
 end
diff --git a/spec/acceptance/broker_api_compatibility/broker_api_v2.13_spec.rb b/spec/acceptance/broker_api_compatibility/broker_api_v2.13_spec.rb
index 1a881a776fc..76d064d19fb 100644
--- a/spec/acceptance/broker_api_compatibility/broker_api_v2.13_spec.rb
+++ b/spec/acceptance/broker_api_compatibility/broker_api_v2.13_spec.rb
@@ -4,7 +4,7 @@
   describe 'v2.13' do
     include VCAP::CloudController::BrokerApiHelper
 
-    let(:create_instance_schema) { {} }
+    let(:create_instance_schema) { { 'type': 'object' } }
     let(:schemas) {
       {
         'service_instance' => {
@@ -26,7 +26,7 @@
     context 'when a broker catalog defines plan schemas' do
       let(:create_instance_schema) {
         {
-          '$schema' => 'http://example.com/schema',
+          '$schema' => 'http://json-schema.org/draft-04/schema#',
           'type' => 'object'
         }
       }
diff --git a/spec/acceptance/service_broker_spec.rb b/spec/acceptance/service_broker_spec.rb
index b80ff965bb3..befa351122e 100644
--- a/spec/acceptance/service_broker_spec.rb
+++ b/spec/acceptance/service_broker_spec.rb
@@ -123,7 +123,14 @@ def build_service(attrs={})
                 {
                   id: 'plan-1',
                   name: 'small',
-                  description: 'A small shared database with 100mb storage quota and 10 connections'
+                  description: 'A small shared database with 100mb storage quota and 10 connections',
+                  schemas: {
+                    service_instance: {
+                      create: {
+                        parameters: { properties: true }
+                      }
+                    }
+                  }
                 }, {
                   id: 'plan-2',
                   name: 'large',
@@ -196,6 +203,11 @@ def build_service(attrs={})
           "Service dashboard_client id must be unique\n" \
           "Service service-1\n" \
           "  Service id must be a string, but has value 12345\n" \
+          "  Plan small\n" \
+          "    Schemas\n" \
+          '      Schema service_instance.create.parameters is not valid. Must conform to JSON Schema Draft 04: '\
+          "The property '#/properties' of type boolean did not match the following type: object in schema "\
+          "http://json-schema.org/draft-04/schema#\n" \
           "Service service-2\n" \
           "  Plan ids must be unique within a service. Service service-2 already has a plan with id 'plan-b'\n" \
           "  Plan large\n" \
diff --git a/spec/request/v2/service_brokers_spec.rb b/spec/request/v2/service_brokers_spec.rb
new file mode 100644
index 00000000000..c0d71aa9f66
--- /dev/null
+++ b/spec/request/v2/service_brokers_spec.rb
@@ -0,0 +1,76 @@
+require 'spec_helper'
+
+RSpec.describe 'ServiceBrokers' do
+  describe 'POST /v2/service_brokers' do
+    service_name = 'myservice'
+    plan_name = 'myplan'
+
+    before do
+      allow(VCAP::Services::ServiceBrokers::V2::Client).to receive(:new) do |*args, **kwargs, &block|
+        fb = FakeServiceBrokerV2Client.new(*args, **kwargs, &block)
+        fb.service_name = service_name
+        fb.plan_name = plan_name
+        fb
+      end
+    end
+
+    it 'should register the service broker' do
+      req_body = {
+        name: 'service-broker-name',
+        broker_url: 'https://broker.example.com',
+        auth_username: 'admin',
+        auth_password: 'secretpassw0rd'
+      }
+
+      post '/v2/service_brokers', req_body.to_json, admin_headers
+      expect(last_response.status).to eq(201)
+
+      broker = VCAP::CloudController::ServiceBroker.last
+      expect(broker.name).to eq(req_body[:name])
+      expect(broker.broker_url).to eq(req_body[:broker_url])
+      expect(broker.auth_username).to eq(req_body[:auth_username])
+      expect(broker.auth_password).to eq(req_body[:auth_password])
+
+      service = VCAP::CloudController::Service.last
+      expect(service.label).to eq(service_name)
+
+      plan = VCAP::CloudController::ServicePlan.last
+      expect(plan.name).to eq(plan_name)
+    end
+
+    context 'for brokers with schemas' do
+      big_string = 'x' * 65 * 1024
+
+      schemas = {
+      'service_instance' => {
+        'create' =>  {
+          'parameters' => {
+              'type' => 'object',
+              'foo' => big_string
+            }
+          }
+        }
+      }
+
+      before do
+        allow(VCAP::Services::ServiceBrokers::V2::Client).to receive(:new) do |*args, **kwargs, &block|
+          fb = FakeServiceBrokerV2Client.new(*args, **kwargs, &block)
+          fb.plan_schemas = schemas
+          fb
+        end
+      end
+
+      it 'should not allow schema bigger than 64KB' do
+        req_body = {
+          name: 'service-broker-name',
+          broker_url: 'https://broker.example.com',
+          auth_username: 'admin',
+          auth_password: 'secretpassw0rd'
+        }
+
+        post '/v2/service_brokers', req_body.to_json, admin_headers
+        expect(last_response.status).to eq(502)
+      end
+    end
+  end
+end
diff --git a/spec/support/fakes/fake_service_broker_v2_client.rb b/spec/support/fakes/fake_service_broker_v2_client.rb
index f2d019104e0..46954860da0 100644
--- a/spec/support/fakes/fake_service_broker_v2_client.rb
+++ b/spec/support/fakes/fake_service_broker_v2_client.rb
@@ -2,24 +2,31 @@ class FakeServiceBrokerV2Client
   attr_accessor :credentials
   attr_accessor :syslog_drain_url
   attr_accessor :volume_mounts
+  attr_accessor :service_name
+  attr_accessor :plan_name
+  attr_accessor :plan_schemas
 
   def initialize(_attrs)
     @credentials = { 'username' => 'cool_user' }
     @syslog_drain_url = 'syslog://drain.example.com'
     @volume_mounts = []
+    @service_name = 'service_name'
+    @plan_name = 'fake_plan_name'
+    @plan_schemas = nil
   end
 
   def catalog
     {
       'services' => [{
         'id'          => 'service_id',
-        'name'        => 'service_name',
+        'name'        => service_name,
         'description' => 'some description',
         'bindable'    => true,
         'plans'       => [{
           'id'          => 'fake_plan_id',
-          'name'        => 'fake_plan_name',
-          'description' => 'fake_plan_description'
+          'name'        => plan_name,
+          'description' => 'fake_plan_description',
+          'schemas'     => plan_schemas
         }]
       }]
     }
diff --git a/spec/unit/lib/services/service_brokers/service_manager_spec.rb b/spec/unit/lib/services/service_brokers/service_manager_spec.rb
index 1ad10d96d6e..affd97b1330 100644
--- a/spec/unit/lib/services/service_brokers/service_manager_spec.rb
+++ b/spec/unit/lib/services/service_brokers/service_manager_spec.rb
@@ -33,7 +33,17 @@ module VCAP::Services::ServiceBrokers
       }
     end
     let(:plan_schemas_hash) do
-      { 'schemas' => { 'service_instance' => { 'create' => { 'parameters' => { '$schema': 'example.com/schema' } } } } }
+      {
+          'schemas' => {
+              'service_instance' => {
+                  'create' => {
+                      'parameters' => {
+                          '$schema' => 'http://json-schema.org/draft-04/schema', 'type' => 'object'
+                      }
+                  }
+              }
+          }
+      }
     end
 
     let(:catalog_hash) do
@@ -161,7 +171,7 @@ module VCAP::Services::ServiceBrokers
           'public' => service_plan.public,
           'active' => service_plan.active,
           'bindable' => true,
-          'create_instance_schema' => '{"$schema":"example.com/schema"}',
+          'create_instance_schema' => '{"$schema":"http://json-schema.org/draft-04/schema","type":"object"}',
         })
       end
 
@@ -196,7 +206,7 @@ module VCAP::Services::ServiceBrokers
           expect(plan.name).to eq(plan_name)
           expect(plan.description).to eq(plan_description)
           expect(JSON.parse(plan.extra)).to eq({ 'cost' => '0.0' })
-          expect(plan.create_instance_schema).to eq('{"$schema":"example.com/schema"}')
+          expect(plan.create_instance_schema).to eq('{"$schema":"http://json-schema.org/draft-04/schema","type":"object"}')
 
           expect(plan.free).to be false
         end
@@ -336,7 +346,7 @@ module VCAP::Services::ServiceBrokers
             expect(plan.description).to eq(plan_description)
             expect(plan.free).to be false
             expect(plan.bindable).to be true
-            expect(plan.create_instance_schema).to eq('{"$schema":"example.com/schema"}')
+            expect(plan.create_instance_schema).to eq('{"$schema":"http://json-schema.org/draft-04/schema","type":"object"}')
           end
 
           it 'creates service audit events for each service plan updated' do
@@ -361,7 +371,7 @@ module VCAP::Services::ServiceBrokers
               'extra' => '{"cost":"0.0"}',
               'bindable' => true,
               'free' => false,
-              'create_instance_schema' => '{"$schema":"example.com/schema"}',
+              'create_instance_schema' => '{"$schema":"http://json-schema.org/draft-04/schema","type":"object"}',
             })
           end
 
diff --git a/spec/unit/lib/services/service_brokers/v2/catalog_schemas_spec.rb b/spec/unit/lib/services/service_brokers/v2/catalog_schemas_spec.rb
index 4ee906cc405..c7145ae2ac9 100644
--- a/spec/unit/lib/services/service_brokers/v2/catalog_schemas_spec.rb
+++ b/spec/unit/lib/services/service_brokers/v2/catalog_schemas_spec.rb
@@ -3,13 +3,11 @@
 module VCAP::Services::ServiceBrokers::V2
   RSpec.describe CatalogSchemas do
     describe 'initializing' do
-      let(:create_instance_schema) { { '$schema': 'example.com/schema' } }
+      let(:path) { 'service_instance.create.parameters' }
+      let(:create_instance_schema) { { '$schema' => 'http://json-schema.org/draft-04/schema#', 'type' => 'object' } }
       let(:attrs) { { 'service_instance' => { 'create' => { 'parameters' => create_instance_schema } } } }
-      subject { CatalogSchemas.new(attrs) }
 
-      its(:create_instance) { should eq create_instance_schema }
-      its(:errors) { should be_empty }
-      its(:valid?) { should be true }
+      subject { CatalogSchemas.new(attrs) }
 
       context 'when attrs have nil value' do
         {
@@ -61,6 +59,170 @@ module VCAP::Services::ServiceBrokers::V2
           end
         end
       end
+
+      context 'when attrs has a potentially dangerous uri' do
+        let(:attrs) {
+          {
+            'service_instance' => {
+              'create' => {
+                'parameters' => 'https://example.com/hax0r'
+              }
+            }
+          }
+        }
+
+        its(:create_instance) { should be_nil }
+        its('errors.messages') { should have(1).items }
+        its('errors.messages.first') { should eq "Schemas #{path} must be a hash, but has value \"https://example.com/hax0r\"" }
+        its(:valid?) { should be false }
+      end
+
+      context 'when the schema does not conform to JSON Schema Draft 04' do
+        let(:create_instance_schema) { { 'properties': true } }
+
+        its(:valid?) { should be false }
+        its('errors.messages') { should have(1).items }
+        its('errors.messages.first') {
+          should match "Schema #{path} is not valid\. Must conform to JSON Schema Draft 04.+properties"
+        }
+      end
+
+      context 'when the schema does not conform to JSON Schema Draft 04 with multiple problems' do
+        let(:create_instance_schema) { { 'type': 'foo', 'properties': true } }
+
+        its(:valid?) { should be false }
+        its('errors.messages') { should have(2).items }
+        its('errors.messages.first') { should match 'properties' }
+        its('errors.messages.second') { should match 'type' }
+      end
+
+      context 'when the schema has an external schema' do
+        let(:create_instance_schema) { { '$schema': 'http://example.com/schema' } }
+
+        its(:valid?) { should be false }
+        its('errors.messages') { should have(1).items }
+        its('errors.messages.first') {
+          should match "Schema #{path} is not valid\. Custom meta schemas are not supported.+http://example.com/schema"
+        }
+      end
+
+      context 'when the schema has an external uri reference' do
+        let(:create_instance_schema) { { '$ref': 'http://example.com/ref' } }
+
+        its(:valid?) { should be false }
+        its('errors.messages') { should have(1).items }
+        its('errors.messages.first') {
+          should match "Schema #{path} is not valid\. No external references are allowed.+http://example.com/ref"
+        }
+      end
+
+      context 'when the schema has an external file reference' do
+        let(:create_instance_schema) { { '$ref': 'path/to/schema.json' } }
+
+        its(:valid?) { should be false }
+        its('errors.messages') { should have(1).items }
+        its('errors.messages.first') {
+          should match "Schema #{path} is not valid\. No external references are allowed.+path/to/schema.json"
+        }
+      end
+
+      context 'when the schema has an internal reference' do
+        let(:create_instance_schema) {
+          {
+            'type' => 'object',
+            'properties': {
+              'foo': { 'type': 'integer' },
+              'bar': { '$ref': '#/properties/foo' }
+            }
+          }
+        }
+
+        its(:valid?) { should be true }
+        its(:errors) { should be_empty }
+      end
+
+      context 'when the schema has an unknown parse error' do
+        before do
+          allow(JSON::Validator).to receive(:validate!) { raise 'some unknown error' }
+        end
+
+        its(:valid?) { should be false }
+        its('errors.messages') { should have(1).items }
+        its('errors.messages.first') { should match "Schema #{path} is not valid\.+ some unknown error" }
+      end
+
+      context 'when the schema has multiple valid constraints ' do
+        let(:create_instance_schema) {
+          { :'$schema' => 'http://json-schema.org/draft-04/schema#',
+            'type' => 'object',
+            :properties => { 'foo': { 'type': 'string' } },
+            :required => ['foo']
+        }
+        }
+        its(:valid?) { should be true }
+        its(:errors) { should be_empty }
+      end
+
+      describe 'schema sizes' do
+        def create_schema_of_size(bytes)
+          surrounding_bytes = 26
+          {
+            'type' => 'object',
+            'foo' => 'x' * (bytes - surrounding_bytes)
+          }
+        end
+
+        context 'that are valid' do
+          {
+            'well below the limit': 1,
+            'just below the limit': 63,
+            'on the limit': 64,
+          }.each do |desc, size_in_kb|
+            context "when the schema is #{desc}" do
+              let(:create_instance_schema) { create_schema_of_size(size_in_kb * 1024) }
+
+              its(:valid?) { should be true }
+              its(:errors) { should be_empty }
+
+              it 'does perform further validation' do
+                expect_any_instance_of(CatalogSchemas).to receive(:validate_metaschema)
+                expect_any_instance_of(CatalogSchemas).to receive(:validate_no_external_references)
+                subject
+              end
+            end
+          end
+        end
+
+        context 'that are invalid' do
+          {
+            'just above the limit': 65,
+            'well above the limit': 10 * 1024,
+          }.each do |desc, size_in_kb|
+            context "when the schema is #{desc}" do
+              path = 'service_instance.create.parameters'
+              let(:create_instance_schema) { create_schema_of_size(size_in_kb * 1024) }
+
+              its(:valid?) { should be false }
+              its('errors.messages') { should have(1).items }
+              its('errors.messages.first') { should match "Schema #{path} is larger than 64KB" }
+
+              it 'does not perform further validation' do
+                expect_any_instance_of(CatalogSchemas).to_not receive(:validate_metaschema)
+                expect_any_instance_of(CatalogSchemas).to_not receive(:validate_no_external_references)
+                subject
+              end
+            end
+          end
+        end
+      end
+
+      context 'when the schema does not have a type field' do
+        let(:create_instance_schema) { { '$schema': 'http://json-schema.org/draft-04/schema#' } }
+
+        its(:valid?) { should be false }
+        its('errors.messages') { should have(1).items }
+        its('errors.messages.first') { should match "Schema #{path} is not valid\.+ must have field \"type\", with value \"object\"" }
+      end
     end
   end
 end