root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

Merge default into bug19200

This commit is contained in:
Hubert Plociniczak 2008-08-14 14:23:01 +01:00
parent 1ac6911c3e 1dd2f6ec7f
commit 26e17f2c24
25 changed files with 390 additions and 974 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
Makefile
View File

@ -119,6 +119,7 @@ srcdist: distclean
cp codegen.py Makefile dist/$(TARBALL_NAME)
cp -r scripts dist/$(TARBALL_NAME)
cp -r docs dist/$(TARBALL_NAME)
chmod 0755 dist/$(TARBALL_NAME)/scripts/*
(cd dist; tar -zcf $(TARBALL_NAME).tar.gz $(TARBALL_NAME))
@ -133,12 +134,20 @@ distclean: clean
install: all
@[ -n "$(TARGET_DIR)" ] || (echo "Please set TARGET_DIR."; false)
@[ -n "$(SBIN_DIR)" ] || (echo "Please set SBIN_DIR."; false)
@[ -n "$(MAN_DIR)" ] || (echo "Please set MAN_DIR."; false)
$(MAKE) VERSION=$(VERSION) GENERIC_STAGE_DIR=$(TARGET_DIR) generic_stage
chmod 0755 scripts/*
mkdir -p $(SBIN_DIR)
mkdir -p $(MAN_DIR)/man1
cp scripts/rabbitmq-server $(SBIN_DIR)
cp scripts/rabbitmqctl $(SBIN_DIR)
cp scripts/rabbitmq-multi $(SBIN_DIR)
for manpage in docs/*.pod ; do \
pod2man -c "RabbitMQ AMQP Server" -d "" -r "" \
$$manpage | gzip --best > \
$(MAN_DIR)/man1/`echo $$manpage | sed -e 's:docs/\(.*\)\.pod:\1\.1\.gz:g'`; \
done
rm -f $(TARGET_DIR)/BUILD

52
docs/rabbitmq-multi.pod Normal file
View File

@ -0,0 +1,52 @@
=head1 NAME
rabbitmq-multi - start/stop local cluster RabbitMQ nodes
=head1 SYNOPSIS
rabbitmq-multi I<command> [command option]
=head1 DESCRIPTION
RabbitMQ is an implementation of AMQP, the emerging standard for high
performance enterprise messaging. The RabbitMQ server is a robust and
scalable implementation of an AMQP broker.
rabbitmq-multi scripts allows for easy set-up of a cluster on a single
machine.
See also rabbitmq-server(1) for configuration information.
=head1 COMMANDS
start_all I<count>
start count nodes with unique names, listening on all IP addresses
and on sequential ports starting from 5672.
stop_all
stop all local RabbitMQ nodes
=head1 EXAMPLES
Start 3 local RabbitMQ nodes with unique, sequential port numbers:
rabbitmq-multi start_all 3
=head1 SEE ALSO
rabbitmq-server(1), rabbitmqctl(1)
=head1 AUTHOR
Originally written by The RabbitMQ Team <info@lshift.net>
=head1 COPYRIGHT
This package, the RabbitMQ server is licensed under the MPL.
If you have any questions regarding licensing, please contact us at
info@rabbitmq.com.
=head1 REFERENCES
RabbitMQ Web Site: http://www.rabbitmq.com

80
docs/rabbitmq-server.pod Normal file
View File

@ -0,0 +1,80 @@
=head1 NAME
rabbitmq-server - start RabbitMQ AMQP server
=head1 SYNOPSIS
rabbitmq-server [-detached]
=head1 DESCRIPTION
RabbitMQ is an implementation of AMQP, the emerging standard for high
performance enterprise messaging. The RabbitMQ server is a robust and
scalable implementation of an AMQP broker.
Running rabbitmq-server in the foreground displays a banner message,
and reports on progress in the startup sequence, concluding with the
message "broker running", indicating that the RabbitMQ broker has been
started successfully. To shut down the server, just terminate the
process or use rabbitmqctl(1).
=head1 ENVIRONMENT
B<MNESIA_BASE>
Defaults to /var/lib/rabbitmq/mnesia. Set this to the directory
where Mnesia database files should be placed.
B<LOG_BASE>
Defaults to /var/log/rabbitmq. Log files generated by the server
will be placed in this directory.
B<NODENAME>
Defaults to rabbit. This can be useful if you want to run more
than one node per machine - B<NODENAME> should be unique per
erlang-node-and-machine combination. See clustering on a single
machine guide
at http://www.rabbitmq.com/clustering.html#single-machine for
details.
B<NODE_IP_ADDRESS>
Defaults to 0.0.0.0. This can be changed if you only want to bind
to one network interface.
B<NODE_PORT>
Defaults to 5672.
B<CLUSTER_CONFIG_FILE>
Defaults to /etc/default/rabbitmq_cluster.config. If this file is
present it is used by the server to auto-configure a RabbitMQ
cluster.
See the clustering guide at http://www.rabbitmq.com/clustering.html
for details.
=head1 OPTIONS
B<-detached> start the server process in the background
=head1 EXAMPLES
Run RabbitMQ AMQP server in the background:
rabbitmq-server -detached
=head1 SEE ALSO
rabbitmq-multi(1), rabbitmqctl(1)
=head1 AUTHOR
Originally written by The RabbitMQ Team <info@lshift.net>
=head1 COPYRIGHT
This package, the RabbitMQ server is licensed under the MPL.
If you have any questions regarding licensing, please contact us at
info@rabbitmq.com.
=head1 REFERENCES
RabbitMQ Web Site: http://www.rabbitmq.com

139
docs/rabbitmqctl.pod Normal file
View File

@ -0,0 +1,139 @@
=head1 NAME
rabbitmqctl - command line tool for managing a RabbitMQ broker
=head1 SYNOPSIS
rabbitmqctl [-n I<node>] I<<command>> [command options]
=head1 DESCRIPTION
RabbitMQ is an implementation of AMQP, the emerging standard for high
performance enterprise messaging. The RabbitMQ server is a robust and
scalable implementation of an AMQP broker.
rabbitmqctl is a command line tool for managing a RabbitMQ broker.
It performs all actions by connecting to one of the broker's nodes.
=head1 OPTIONS
B<-n> I<node>
default node is C<rabbit@server>, where server is the local host.
On a host named C<server.example.com>, the node name of the RabbitMQ
Erlang node will usually be rabbit@server (unless NODENAME has been
set to some non-default value at broker startup time).
The output of hostname -s is usually the correct suffix to use
after the "@" sign. See rabbitmq-server(1) for details of configur-
ing the RabbitMQ broker.
=head1 COMMANDS
=head2 APPLICATION AND CLUSTER MANAGEMENT
stop
stop the Erlang node on which RabbitMQ broker is running.
stop_app
stop the RabbitMQ application, leaving the Erlang node running.
This command is typically run prior to performing other management
actions that require the RabbitMQ application to be stopped,
e.g. I<reset>.
start_app
start the RabbitMQ application.
This command is typically run prior to performing other management
actions that require the RabbitMQ application to be stopped,
e.g. I<reset>.
status
display various information about the RabbitMQ broker, such as
whether the RabbitMQ application on the current node, its version
number, what nodes are part of the broker, which of these are
running.
force
return a RabbitMQ node to its virgin state.
Removes the node from any cluster it belongs to, removes all data
from the management database, such as configured users, vhosts and
deletes all persistent messages.
force_reset
the same as I<force> command, but resets the node unconditionally,
regardless of the current management database state and cluster
configuration.
It should only be used as a last resort if the database or cluster
configuration has been corrupted.
cluster I<clusternode> ...
instruct the node to become member of a cluster with the specified
nodes determined by I<clusternode> option(s).
See http://www.rabbitmq.com/clustering.html for more information
about clustering.
=head2 USER MANAGEMENT
add_user I<username> I<password>
create a user named I<username> with (initial) password I<password>.
change_password I<username> I<newpassword>
change the password for the user named I<username> to I<newpassword>.
list_users
list all users.
=head2 ACCESS CONTROL
add_vhost I<vhostpath>
create a new virtual host called I<vhostpath>.
delete_vhost I<vhostpath>
delete a virtual host I<vhostpath>.
That command deletes also all its exchanges, queues and user mappings.
list_vhosts
list all virtual hosts.
map_user_vhost I<username> I<vhostpath>
grant the user named I<username> access to the virtual host called
I<vhostpath>.
unmap_user_vhost I<username> I<vhostpath>
deny the user named I<username> access to the virtual host called
I<vhostpath>.
list_user_vhost I<username>
list all the virtual hosts to which the user named I<username> has
been granted access.
=head1 EXAMPLES
Create a user named foo with (initial) password bar at the Erlang node
rabbit@test:
rabbitmqctl -n rabbit@test add_user foo bar
Grant user named foo access to the virtual host called test at the
default Erlang node:
rabbitmqctl map_user_vhost foo test
=head1 SEE ALSO
rabbitmq-multi(1), rabbitmq-server(1)
=head1 AUTHOR
Originally written by The RabbitMQ Team <info@lshift.net>
=head1 COPYRIGHT
This package, the RabbitMQ server is licensed under the MPL.
If you have any questions regarding licensing, please contact us at
info@rabbitmq.com.
=head1 REFERENCES
RabbitMQ Web Site: http://www.rabbitmq.com

2
ebin/rabbit.app
View File

@ -25,11 +25,9 @@
rabbit_node_monitor,
rabbit_persister,
rabbit_reader,
rabbit_realm,
rabbit_router,
rabbit_sup,
rabbit_tests,
rabbit_ticket,
rabbit_tracer,
rabbit_writer,
tcp_acceptor,

32
include/rabbit.hrl
View File

@ -27,28 +27,20 @@
-record(user_vhost, {username, virtual_host}).
-record(vhost, {virtual_host, dummy}).
-record(vhost_realm, {virtual_host, realm}).
-record(realm, {name, ignore}).
-record(realm_resource, {realm, resource}).
-record(user_realm, {username, realm, ticket_pattern}).
-record(realm_visitor, {realm, pid}).
-record(connection, {user, timeout_sec, frame_max, vhost}).
-record(content, {class_id,
properties, %% either 'none', or a decoded record/tuple
properties_bin, %% either 'none', or an encoded properties binary
%% Note: at most one of properties and properties_bin can be 'none' at once.
payload_fragments_rev %% list of binaries, in reverse order (!)
}).
-record(content,
{class_id,
properties, %% either 'none', or a decoded record/tuple
properties_bin, %% either 'none', or an encoded properties binary
%% Note: at most one of properties and properties_bin can be
%% 'none' at once.
payload_fragments_rev %% list of binaries, in reverse order (!)
}).
-record(resource, {virtual_host, kind, name}).
-record(ticket, {realm_name, passive_flag, active_flag, write_flag, read_flag}).
-record(exchange, {name, type, durable, auto_delete, arguments}).
-record(amqqueue, {name, durable, auto_delete, arguments, binding_specs, pid}).
@ -80,19 +72,11 @@
#resource{virtual_host :: vhost(),
kind :: Kind,
name :: name()}).
-type(realm_name() :: r('realm')).
-type(queue_name() :: r('queue')).
-type(exchange_name() :: r('exchange')).
-type(user() ::
#user{username :: username(),
password :: password()}).
-type(ticket() ::
#ticket{realm_name :: realm_name(),
passive_flag :: bool(),
active_flag :: bool(),
write_flag :: bool(),
read_flag :: bool()}).
-type(permission() :: 'passive' | 'active' | 'write' | 'read').
-type(binding_spec() ::
#binding_spec{exchange_name :: exchange_name(),
routing_key :: routing_key(),

13
packaging/RPMS/Fedora/rabbitmq-server.spec
View File

@ -20,6 +20,7 @@ scalable implementation of an AMQP broker.
%define _libdir /usr/lib/erlang
%define _docdir /usr/share/doc
%define _mandir /usr/share/man
%define _maindir $RPM_BUILD_ROOT%{_libdir}/lib/rabbitmq_server-%{main_version}
%define package_name rabbitmq-server-dist
@ -36,8 +37,10 @@ fi
%build
mkdir %{package_name}
mkdir %{package_name}/sbin
mkdir %{package_name}/man
make install TARGET_DIR=`pwd`/%{package_name} \
SBIN_DIR=`pwd`/%{package_name}/sbin \
MAN_DIR=`pwd`/%{package_name}/man
VERSION=%{main_version}
%install
@ -45,6 +48,7 @@ mkdir -p %{_maindir}
mkdir -p $RPM_BUILD_ROOT%{_docdir}/rabbitmq-server
mkdir -p $RPM_BUILD_ROOT/etc/init.d
mkdir -p $RPM_BUILD_ROOT/usr/sbin
mkdir -p $RPM_BUILD_ROOT%{_mandir}
mkdir -p $RPM_BUILD_ROOT/var/lib/rabbitmq/mnesia
mkdir -p $RPM_BUILD_ROOT/var/log/rabbitmq
@ -55,6 +59,7 @@ cp -r %{package_name}/src %{_maindir}
cp -r %{package_name}/include %{_maindir}
chmod 755 %{package_name}/sbin/*
cp %{package_name}/sbin/* $RPM_BUILD_ROOT/usr/sbin/
cp -r %{package_name}/man/* $RPM_BUILD_ROOT%{_mandir}/
cp ../init.d $RPM_BUILD_ROOT/etc/init.d/rabbitmq-server
chmod 775 $RPM_BUILD_ROOT/etc/init.d/rabbitmq-server
@ -63,6 +68,8 @@ mv $RPM_BUILD_ROOT/usr/sbin/rabbitmqctl $RPM_BUILD_ROOT/usr/sbin/rabbitmqctl_rea
cp ../rabbitmqctl_wrapper $RPM_BUILD_ROOT/usr/sbin/rabbitmqctl
chmod 755 $RPM_BUILD_ROOT/usr/sbin/rabbitmqctl
cp %{buildroot}%{_mandir}/man1/rabbitmqctl.1.gz %{buildroot}%{_mandir}/man1/rabbitmqctl_real.1.gz
%post
# create rabbitmq group
if ! getent group rabbitmq >/dev/null; then
@ -107,10 +114,8 @@ fi
%defattr(-,root,root)
%{_libdir}/lib/rabbitmq_server-%{main_version}/
%{_docdir}/rabbitmq-server/
/usr/sbin/rabbitmq-server
/usr/sbin/rabbitmq-multi
/usr/sbin/rabbitmqctl
/usr/sbin/rabbitmqctl_real
%{_mandir}
/usr/sbin
/var/lib/rabbitmq
/var/log/rabbitmq
/etc/init.d/rabbitmq-server

2
packaging/debs/Debian/debian/control
View File

@ -12,4 +12,4 @@ Description: An AMQP server written in Erlang
RabbitMQ is an implementation of AMQP, the emerging standard for high
performance enterprise messaging. The RabbitMQ server is a robust and
scalable implementation of an AMQP broker.
Homepage: http://www.rabbitmq.com/
Homepage: http://www.rabbitmq.com/

2
packaging/debs/Debian/debian/dirs
View File

@ -1,6 +1,6 @@
usr/lib/erlang/lib
usr/sbin
usr/share/linda/overrides
usr/share/man
var/lib/rabbitmq/mnesia
var/log/rabbitmq

4
packaging/debs/Debian/debian/rules
View File

@ -5,7 +5,7 @@ include /usr/share/cdbs/1/class/makefile.mk
RABBIT_LIB=$(DEB_DESTDIR)usr/lib/erlang/lib/rabbitmq_server-$(DEB_UPSTREAM_VERSION)
DEB_MAKE_INSTALL_TARGET := install TARGET_DIR=$(RABBIT_LIB)/ SBIN_DIR=$(DEB_DESTDIR)usr/sbin
DEB_MAKE_INSTALL_TARGET := install TARGET_DIR=$(RABBIT_LIB)/ SBIN_DIR=$(DEB_DESTDIR)usr/sbin MAN_DIR=$(DEB_DESTDIR)usr/share/man
DOCDIR=$(DEB_DESTDIR)usr/share/doc/rabbitmq-server/
@ -14,5 +14,5 @@ install/rabbitmq-server::
rm $(RABBIT_LIB)/LICENSE*
mv $(DEB_DESTDIR)usr/sbin/rabbitmqctl $(DEB_DESTDIR)usr/sbin/rabbitmqctl_real
cp debian/rabbitmqctl_wrapper $(DEB_DESTDIR)usr/sbin/rabbitmqctl
cp $(DEB_DESTDIR)usr/share/man/man1/rabbitmqctl.1.gz $(DEB_DESTDIR)usr/share/man/man1/rabbitmqctl_real.1.gz
chmod a+x $(DEB_DESTDIR)usr/sbin/rabbitmqctl
echo "Tag: usr-lib-in-arch-all" > $(DEB_DESTDIR)usr/share/linda/overrides/rabbitmq-server

3
packaging/generic-unix/Makefile
View File

@ -7,11 +7,10 @@ dist:
make -C ../.. VERSION=$(VERSION) srcdist
tar -zxvf ../../dist/$(SOURCE_DIR).tar.gz
mkdir $(TARGET_DIR)
mkdir $(TARGET_DIR)/sbin
make -C $(SOURCE_DIR) \
TARGET_DIR=`pwd`/$(TARGET_DIR) \
SBIN_DIR=`pwd`/$(TARGET_DIR)/sbin \
MAN_DIR=`pwd`/$(TARGET_DIR)/share/man \
install
tar -zcf $(TARGET_TARBALL).tar.gz $(TARGET_DIR)

1
packaging/windows/Makefile
View File

@ -15,6 +15,7 @@ dist:
rm -rf $(SOURCE_DIR)/scripts
rm -rf $(SOURCE_DIR)/codegen* $(SOURCE_DIR)/Makefile
rm -f $(SOURCE_DIR)/BUILD
rm -rf $(SOURCE_DIR)/docs
mv $(SOURCE_DIR) $(TARGET_DIR)
zip -r $(TARGET_ZIP).zip $(TARGET_DIR)

38
src/rabbit.erl
View File

@ -103,7 +103,7 @@ manage_applications(Iterate, Do, Undo, SkipError, ErrorTag, Apps) ->
end
end, [], Apps),
ok.
start_applications(Apps) ->
manage_applications(fun lists:foldl/3,
fun application:start/1,
@ -133,9 +133,9 @@ start(normal, []) ->
io:format("starting ~-20s ...", [Msg]),
Thunk(),
io:format("done~n");
({Msg, M, F, A}) ->
({Msg, M, F, A}) ->
io:format("starting ~-20s ...", [Msg]),
apply(M, F, A),
apply(M, F, A),
io:format("done~n")
end,
[{"database",
@ -155,14 +155,12 @@ start(normal, []) ->
{"recovery",
fun () ->
ok = maybe_insert_default_data(),
ok = rabbit_exchange:recover(),
ok = rabbit_amqqueue:recover(),
ok = rabbit_realm:recover()
ok = rabbit_amqqueue:recover()
end},
{"persister",
fun () ->
ok = start_child(rabbit_persister)
fun () ->
ok = start_child(rabbit_persister)
end},
{"builtin applications",
fun () ->
@ -220,26 +218,8 @@ insert_default_data() ->
{ok, DefaultPass} = application:get_env(default_pass),
{ok, DefaultVHost} = application:get_env(default_vhost),
ok = rabbit_access_control:add_vhost(DefaultVHost),
ok = insert_default_user(DefaultUser, DefaultPass,
[{DefaultVHost, [<<"/data">>, <<"/admin">>]}]),
ok.
insert_default_user(Username, Password, VHostSpecs) ->
ok = rabbit_access_control:add_user(Username, Password),
lists:foreach(
fun ({VHostPath, Realms}) ->
ok = rabbit_access_control:map_user_vhost(
Username, VHostPath),
lists:foreach(
fun (Realm) ->
RealmFullName =
rabbit_misc:r(VHostPath, realm, Realm),
ok = rabbit_access_control:map_user_realm(
Username,
rabbit_access_control:full_ticket(
RealmFullName))
end, Realms)
end, VHostSpecs),
ok = rabbit_access_control:add_user(DefaultUser, DefaultPass),
ok = rabbit_access_control:map_user_vhost(DefaultUser, DefaultVHost),
ok.
start_builtin_amq_applications() ->
@ -278,7 +258,7 @@ error_log_location() ->
end.
sasl_log_location() ->
case application:get_env(sasl, sasl_error_logger) of
case application:get_env(sasl, sasl_error_logger) of
{ok, {file, File}} -> File;
{ok, false} -> undefined;
{ok, tty} -> tty;

124
src/rabbit_access_control.erl
View File

@ -28,12 +28,11 @@
-include("rabbit.hrl").
-export([check_login/2, user_pass_login/2,
check_vhost_access/2, lookup_realm_access/2]).
check_vhost_access/2]).
-export([add_user/2, delete_user/1, change_password/2, list_users/0,
lookup_user/1]).
-export([add_vhost/1, delete_vhost/1, list_vhosts/0, list_vhost_users/1]).
-export([list_user_vhosts/1, map_user_vhost/2, unmap_user_vhost/2]).
-export([list_user_realms/2, map_user_realm/2, full_ticket/1]).
%%----------------------------------------------------------------------------
@ -42,7 +41,6 @@
-spec(check_login/2 :: (binary(), binary()) -> user()).
-spec(user_pass_login/2 :: (username(), password()) -> user()).
-spec(check_vhost_access/2 :: (user(), vhost()) -> 'ok').
-spec(lookup_realm_access/2 :: (user(), realm_name()) -> maybe(ticket())).
-spec(add_user/2 :: (username(), password()) -> 'ok').
-spec(delete_user/1 :: (username()) -> 'ok').
-spec(change_password/2 :: (username(), password()) -> 'ok').
@ -55,9 +53,6 @@
-spec(list_user_vhosts/1 :: (username()) -> [vhost()]).
-spec(map_user_vhost/2 :: (username(), vhost()) -> 'ok').
-spec(unmap_user_vhost/2 :: (username(), vhost()) -> 'ok').
-spec(map_user_realm/2 :: (username(), ticket()) -> 'ok').
-spec(list_user_realms/2 :: (username(), vhost()) -> [{name(), ticket()}]).
-spec(full_ticket/1 :: (realm_name()) -> ticket()).
-endif.
@ -87,7 +82,7 @@ check_login(<<"AMQPLAIN">>, Response) ->
[LoginTable])
end;
check_login(Mechanism, _Response) ->
check_login(Mechanism, _Response) ->
rabbit_misc:protocol_error(
access_refused, "unsupported authentication mechanism '~s'",
[Mechanism]).
@ -130,18 +125,6 @@ check_vhost_access(#user{username = Username}, VHostPath) ->
[VHostPath, Username])
end.
lookup_realm_access(#user{username = Username}, RealmName = #resource{kind = realm}) ->
%% TODO: use dirty ops instead
rabbit_misc:execute_mnesia_transaction(
fun () ->
case user_realms(Username, RealmName) of
[] ->
none;
[#user_realm{ticket_pattern = TicketPattern}] ->
TicketPattern
end
end).
add_user(Username, Password) ->
R = rabbit_misc:execute_mnesia_transaction(
fun () ->
@ -162,8 +145,7 @@ delete_user(Username) ->
Username,
fun () ->
ok = mnesia:delete({user, Username}),
ok = mnesia:delete({user_vhost, Username}),
ok = mnesia:delete({user_realm, Username})
ok = mnesia:delete({user_vhost, Username})
end)),
rabbit_log:info("Deleted user ~p~n", [Username]),
R.
@ -191,24 +173,14 @@ add_vhost(VHostPath) ->
case mnesia:read({vhost, VHostPath}) of
[] ->
ok = mnesia:write(#vhost{virtual_host = VHostPath}),
DataRealm =
rabbit_misc:r(VHostPath, realm, <<"/data">>),
AdminRealm =
rabbit_misc:r(VHostPath, realm, <<"/admin">>),
ok = rabbit_realm:add_realm(DataRealm),
ok = rabbit_realm:add_realm(AdminRealm),
#exchange{} = rabbit_exchange:declare(
DataRealm, <<"">>,
direct, true, false, []),
#exchange{} = rabbit_exchange:declare(
DataRealm, <<"amq.direct">>,
direct, true, false, []),
#exchange{} = rabbit_exchange:declare(
DataRealm, <<"amq.topic">>,
topic, true, false, []),
#exchange{} = rabbit_exchange:declare(
DataRealm, <<"amq.fanout">>,
fanout, true, false, []),
[rabbit_exchange:declare(
rabbit_misc:r(VHostPath, exchange, Name),
Type, true, false, []) ||
{Name,Type} <-
[{<<"">>, direct},
{<<"amq.direct">>, direct},
{<<"amq.topic">>, topic},
{<<"amq.fanout">>, fanout}]],
ok;
[_] ->
mnesia:abort({vhost_already_exists, VHostPath})
@ -240,11 +212,6 @@ internal_delete_vhost(VHostPath) ->
ok = rabbit_exchange:delete(Name, false)
end,
rabbit_exchange:list_vhost_exchanges(VHostPath)),
lists:foreach(fun (RealmName) ->
ok = rabbit_realm:delete_realm(
rabbit_misc:r(VHostPath, realm, RealmName))
end,
rabbit_realm:list_vhost_realms(VHostPath)),
lists:foreach(fun (Username) ->
ok = unmap_user_vhost(Username, VHostPath)
end,
@ -290,77 +257,8 @@ unmap_user_vhost(Username, VHostPath) ->
rabbit_misc:with_user_and_vhost(
Username, VHostPath,
fun () ->
lists:foreach(fun mnesia:delete_object/1,
user_realms(Username,
rabbit_misc:r(VHostPath, realm))),
ok = mnesia:delete_object(
#user_vhost{username = Username,
virtual_host = VHostPath})
end)).
map_user_realm(Username,
Ticket = #ticket{realm_name = RealmName =
#resource{virtual_host = VHostPath,
kind = realm}}) ->
rabbit_misc:execute_mnesia_transaction(
rabbit_misc:with_user_and_vhost(
Username, VHostPath,
rabbit_misc:with_realm(
RealmName,
fun () ->
lists:foreach(fun mnesia:delete_object/1,
user_realms(Username, RealmName)),
case internal_lookup_vhost_access(Username, VHostPath) of
{ok, _R} ->
case ticket_liveness(Ticket) of
alive ->
ok = mnesia:write(
#user_realm{username = Username,
realm = RealmName,
ticket_pattern = Ticket});
dead ->
ok
end;
not_found ->
mnesia:abort(not_mapped_to_vhost)
end
end))).
list_user_realms(Username, VHostPath) ->
[{Name, Pattern} ||
#user_realm{realm = #resource{name = Name},
ticket_pattern = Pattern} <-
%% TODO: use dirty ops instead
rabbit_misc:execute_mnesia_transaction(
rabbit_misc:with_user_and_vhost(
Username, VHostPath,
fun () ->
case internal_lookup_vhost_access(
Username, VHostPath) of
{ok, _R} ->
user_realms(Username,
rabbit_misc:r(VHostPath, realm));
not_found ->
mnesia:abort(not_mapped_to_vhost)
end
end))].
ticket_liveness(#ticket{passive_flag = false,
active_flag = false,
write_flag = false,
read_flag = false}) ->
dead;
ticket_liveness(_) ->
alive.
full_ticket(RealmName) ->
#ticket{realm_name = RealmName,
passive_flag = true,
active_flag = true,
write_flag = true,
read_flag = true}.
user_realms(Username, RealmName) ->
mnesia:match_object(#user_realm{username = Username,
realm = RealmName,
_ = '_'}).

36
src/rabbit_amqqueue.erl
View File

@ -25,8 +25,8 @@
-module(rabbit_amqqueue).
-export([start/0, recover/0, declare/5, delete/3, purge/1, internal_delete/1]).
-export([pseudo_queue/3]).
-export([start/0, recover/0, declare/4, delete/3, purge/1, internal_delete/1]).
-export([pseudo_queue/2]).
-export([lookup/1, with/2, with_or_die/2, list_vhost_queues/1,
stat/1, stat_all/0, deliver/5, redeliver/2, requeue/3, ack/4,
commit/2, rollback/2]).
@ -55,7 +55,7 @@
{'error', 'queue_not_found' | 'exchange_not_found'}).
-spec(start/0 :: () -> 'ok').
-spec(recover/0 :: () -> 'ok').
-spec(declare/5 :: (realm_name(), name(), bool(), bool(), amqp_table()) ->
-spec(declare/4 :: (queue_name(), bool(), bool(), amqp_table()) ->
amqqueue()).
-spec(add_binding/4 ::
(queue_name(), exchange_name(), routing_key(), amqp_table()) ->
@ -96,7 +96,7 @@
-spec(notify_sent/2 :: (pid(), pid()) -> 'ok').
-spec(internal_delete/1 :: (queue_name()) -> 'ok' | not_found()).
-spec(on_node_down/1 :: (node()) -> 'ok').
-spec(pseudo_queue/3 :: (realm_name(), binary(), pid()) -> amqqueue()).
-spec(pseudo_queue/2 :: (binary(), pid()) -> amqqueue()).
-endif.
@ -130,9 +130,8 @@ recover_durable_queues() ->
ok
end).
declare(RealmName, NameBin, Durable, AutoDelete, Args) ->
QName = rabbit_misc:r(RealmName, queue, NameBin),
Q = start_queue_process(#amqqueue{name = QName,
declare(QueueName, Durable, AutoDelete, Args) ->
Q = start_queue_process(#amqqueue{name = QueueName,
durable = Durable,
auto_delete = AutoDelete,
arguments = Args,
@ -140,9 +139,8 @@ declare(RealmName, NameBin, Durable, AutoDelete, Args) ->
pid = none}),
case rabbit_misc:execute_mnesia_transaction(
fun () ->
case mnesia:wread({amqqueue, QName}) of
case mnesia:wread({amqqueue, QueueName}) of
[] -> ok = recover_queue(Q),
ok = rabbit_realm:add(RealmName, QName),
Q;
[ExistingQ] -> ExistingQ
end
@ -251,7 +249,7 @@ with(Name, F, E) ->
end.
with(Name, F) ->
with(Name, F, fun () -> {error, not_found} end).
with(Name, F, fun () -> {error, not_found} end).
with_or_die(Name, F) ->
with(Name, F, fun () -> rabbit_misc:protocol_error(
not_found, "no ~s", [rabbit_misc:rs(Name)])
@ -338,28 +336,20 @@ internal_delete(QueueName) ->
case mnesia:wread({amqqueue, QueueName}) of
[] -> {error, not_found};
[Q] ->
ok = delete_temp(Q),
ok = delete_queue(Q),
ok = mnesia:delete({durable_queues, QueueName}),
ok = rabbit_realm:delete_from_all(QueueName),
ok
end
end).
delete_temp(Q = #amqqueue{name = QueueName}) ->
delete_queue(Q = #amqqueue{name = QueueName}) ->
ok = delete_bindings(Q),
ok = rabbit_exchange:delete_binding(
default_binding_spec(QueueName), Q),
ok = mnesia:delete({amqqueue, QueueName}),
ok.
delete_queue(Q = #amqqueue{name = QueueName, durable = Durable}) ->
ok = delete_temp(Q),
if
Durable -> ok;
true -> ok = rabbit_realm:delete_from_all(QueueName)
end.
on_node_down(Node) ->
on_node_down(Node) ->
rabbit_misc:execute_mnesia_transaction(
fun () ->
qlc:fold(
@ -370,8 +360,8 @@ on_node_down(Node) ->
node(Pid) == Node]))
end).
pseudo_queue(RealmName, NameBin, Pid) ->
#amqqueue{name = rabbit_misc:r(RealmName, queue, NameBin),
pseudo_queue(QueueName, Pid) ->
#amqqueue{name = QueueName,
durable = false,
auto_delete = false,
arguments = [],

126
src/rabbit_channel.erl
View File

@ -37,7 +37,7 @@
transaction_id, tx_participants, next_tag,
uncommitted_ack_q, unacked_message_q,
username, virtual_host,
most_recently_declared_queue, consumer_mapping, next_ticket}).
most_recently_declared_queue, consumer_mapping}).
%%----------------------------------------------------------------------------
@ -94,8 +94,7 @@ init(ProxyPid, [ReaderPid, WriterPid, Username, VHost]) ->
username = Username,
virtual_host = VHost,
most_recently_declared_queue = <<>>,
consumer_mapping = dict:new(),
next_ticket = 101}.
consumer_mapping = dict:new()}.
handle_message({method, Method, Content}, State) ->
case (catch handle_method(Method, Content, State)) of
@ -140,7 +139,6 @@ handle_message(Other, State) ->
terminate(Reason, State = #ch{writer_pid = WriterPid}) ->
Res = notify_queues(internal_rollback(State)),
ok = rabbit_realm:leave_realms(self()),
case Reason of
normal -> ok = Res;
_ -> ok
@ -195,14 +193,6 @@ die_precondition_failed(Fmt, Params) ->
rabbit_misc:protocol_error({false, 406, <<"PRECONDITION_FAILED">>},
Fmt, Params).
check_ticket(TicketNumber, FieldIndex, Name, #ch{ username = Username}) ->
rabbit_ticket:check_ticket(TicketNumber, FieldIndex, Name, Username).
lookup_ticket(TicketNumber, FieldIndex,
#ch{ username = Username, virtual_host = VHostPath }) ->
rabbit_ticket:lookup_ticket(TicketNumber, FieldIndex,
Username, VHostPath).
%% check that an exchange/queue name does not contain the reserved
%% "amq." prefix.
%%
@ -235,57 +225,19 @@ handle_method(_Method, _, #ch{state = starting}) ->
handle_method(#'channel.close'{}, _, State = #ch{writer_pid = WriterPid}) ->
ok = notify_queues(internal_rollback(State)),
ok = rabbit_realm:leave_realms(self()),
ok = rabbit_writer:send_command(WriterPid, #'channel.close_ok'{}),
ok = rabbit_writer:shutdown(WriterPid),
stop;
handle_method(#'access.request'{realm = RealmNameBin,
exclusive = Exclusive,
passive = Passive,
active = Active,
write = Write,
read = Read},
_, State = #ch{username = Username,
virtual_host = VHostPath,
next_ticket = NextTicket}) ->
RealmName = rabbit_misc:r(VHostPath, realm, RealmNameBin),
Ticket = #ticket{realm_name = RealmName,
passive_flag = Passive,
active_flag = Active,
write_flag = Write,
read_flag = Read},
case rabbit_realm:access_request(Username, Exclusive, Ticket) of
ok ->
rabbit_ticket:record_ticket(NextTicket, Ticket),
NewState = State#ch{next_ticket = NextTicket + 1},
{reply, #'access.request_ok'{ticket = NextTicket}, NewState};
{error, not_found} ->
rabbit_misc:protocol_error(
invalid_path, "no ~s", [rabbit_misc:rs(RealmName)]);
{error, bad_realm_path} ->
%% FIXME: spec bug? access_refused is a soft error, spec requires it to be hard
rabbit_misc:protocol_error(
access_refused, "bad path for ~s", [rabbit_misc:rs(RealmName)]);
{error, resource_locked} ->
rabbit_misc:protocol_error(
resource_locked, "~s is locked", [rabbit_misc:rs(RealmName)]);
{error, access_refused} ->
rabbit_misc:protocol_error(
access_refused,
"~w permissions denied for user '~s' attempting to access ~s",
[rabbit_misc:permission_list(Ticket),
Username, rabbit_misc:rs(RealmName)])
end;
handle_method(#'access.request'{},_, State) ->
{reply, #'access.request_ok'{ticket = 1}, State};
handle_method(#'basic.publish'{ticket = TicketNumber,
exchange = ExchangeNameBin,
handle_method(#'basic.publish'{exchange = ExchangeNameBin,
routing_key = RoutingKey,
mandatory = Mandatory,
immediate = Immediate},
Content, State = #ch{ virtual_host = VHostPath}) ->
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
check_ticket(TicketNumber, #ticket.write_flag, ExchangeName, State),
Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
%% We decode the content's properties here because we're almost
%% certain to want to look at delivery-mode and priority.
@ -323,13 +275,11 @@ handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
uncommitted_ack_q = NewUAQ})
end};
handle_method(#'basic.get'{ticket = TicketNumber,
queue = QueueNameBin,
handle_method(#'basic.get'{queue = QueueNameBin,
no_ack = NoAck},
_, State = #ch{ proxy_pid = ProxyPid, writer_pid = WriterPid,
next_tag = DeliveryTag }) ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
check_ticket(TicketNumber, #ticket.read_flag, QueueName, State),
case rabbit_amqqueue:with_or_die(
QueueName,
fun (Q) -> rabbit_amqqueue:basic_get(Q, ProxyPid, NoAck) end) of
@ -352,8 +302,7 @@ handle_method(#'basic.get'{ticket = TicketNumber,
{reply, #'basic.get_empty'{cluster_id = <<>>}, State}
end;
handle_method(#'basic.consume'{ticket = TicketNumber,
queue = QueueNameBin,
handle_method(#'basic.consume'{queue = QueueNameBin,
consumer_tag = ConsumerTag,
no_local = _, % FIXME: implement
no_ack = NoAck,
@ -365,7 +314,6 @@ handle_method(#'basic.consume'{ticket = TicketNumber,
case dict:find(ConsumerTag, ConsumerMapping) of
error ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
check_ticket(TicketNumber, #ticket.read_flag, QueueName, State),
ActualConsumerTag =
case ConsumerTag of
<<>> -> rabbit_misc:binstring_guid("amq.ctag");
@ -391,7 +339,7 @@ handle_method(#'basic.consume'{ticket = TicketNumber,
ConsumerMapping)}};
{error, queue_owned_by_another_connection} ->
%% The spec is silent on which exception to use
%% here. This seems reasonable?
%% here. This seems reasonable?
%% FIXME: check this
rabbit_misc:protocol_error(
@ -495,8 +443,7 @@ handle_method(#'basic.recover'{}, _, _State) ->
rabbit_misc:protocol_error(
not_allowed, "attempt to recover a transactional channel",[]);
handle_method(#'exchange.declare'{ticket = TicketNumber,
exchange = ExchangeNameBin,
handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
type = TypeNameBin,
passive = false,
durable = Durable,
@ -505,17 +452,13 @@ handle_method(#'exchange.declare'{ticket = TicketNumber,
nowait = NoWait,
arguments = Args},
_, State = #ch{ virtual_host = VHostPath }) ->
#ticket{realm_name = RealmName} =
lookup_ticket(TicketNumber, #ticket.active_flag, State),
CheckedType = rabbit_exchange:check_type(TypeNameBin),
%% FIXME: clarify spec as per declare wrt differing realms
X = case rabbit_exchange:lookup(
rabbit_misc:r(VHostPath, exchange, ExchangeNameBin)) of
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
X = case rabbit_exchange:lookup(ExchangeName) of
{ok, FoundX} -> FoundX;
{error, not_found} ->
ActualNameBin = check_name('exchange', ExchangeNameBin),
rabbit_exchange:declare(RealmName,
ActualNameBin,
check_name('exchange', ExchangeNameBin),
rabbit_exchange:declare(ExchangeName,
CheckedType,
Durable,
AutoDelete,
@ -524,26 +467,21 @@ handle_method(#'exchange.declare'{ticket = TicketNumber,
ok = rabbit_exchange:assert_type(X, CheckedType),
return_ok(State, NoWait, #'exchange.declare_ok'{});
handle_method(#'exchange.declare'{ticket = TicketNumber,
exchange = ExchangeNameBin,
handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
type = TypeNameBin,
passive = true,
nowait = NoWait},
_, State = #ch{ virtual_host = VHostPath }) ->
%% FIXME: spec issue: permit active_flag here as well as passive_flag?
#ticket{} = lookup_ticket(TicketNumber, #ticket.passive_flag, State),
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
X = rabbit_exchange:lookup_or_die(ExchangeName),
ok = rabbit_exchange:assert_type(X, rabbit_exchange:check_type(TypeNameBin)),
return_ok(State, NoWait, #'exchange.declare_ok'{});
handle_method(#'exchange.delete'{ticket = TicketNumber,
exchange = ExchangeNameBin,
handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
if_unused = IfUnused,
nowait = NoWait},
_, State = #ch { virtual_host = VHostPath }) ->
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
check_ticket(TicketNumber, #ticket.active_flag, ExchangeName, State),
case rabbit_exchange:delete(ExchangeName, IfUnused) of
{error, not_found} ->
rabbit_misc:protocol_error(
@ -555,8 +493,7 @@ handle_method(#'exchange.delete'{ticket = TicketNumber,
return_ok(State, NoWait, #'exchange.delete_ok'{})
end;
handle_method(#'queue.declare'{ticket = TicketNumber,
queue = QueueNameBin,
handle_method(#'queue.declare'{queue = QueueNameBin,
passive = false,
durable = Durable,
exclusive = ExclusiveDeclare,
@ -565,8 +502,6 @@ handle_method(#'queue.declare'{ticket = TicketNumber,
arguments = Args},
_, State = #ch { virtual_host = VHostPath,
reader_pid = ReaderPid }) ->
#ticket{realm_name = RealmName} =
lookup_ticket(TicketNumber, #ticket.active_flag, State),
%% FIXME: atomic create&claim
Finish =
fun (Q) ->
@ -587,7 +522,6 @@ handle_method(#'queue.declare'{ticket = TicketNumber,
end,
Q
end,
%% FIXME: clarify spec as per declare wrt differing realms
Q = case rabbit_amqqueue:with(
rabbit_misc:r(VHostPath, queue, QueueNameBin),
Finish) of
@ -597,34 +531,28 @@ handle_method(#'queue.declare'{ticket = TicketNumber,
<<>> -> rabbit_misc:binstring_guid("amq.gen");
Other -> check_name('queue', Other)
end,
Finish(rabbit_amqqueue:declare(RealmName,
ActualNameBin,
Durable,
AutoDelete,
Args));
QueueName = rabbit_misc:r(VHostPath, queue, ActualNameBin),
Finish(rabbit_amqqueue:declare(QueueName,
Durable, AutoDelete, Args));
Other -> Other
end,
return_queue_declare_ok(State, NoWait, Q);
handle_method(#'queue.declare'{ticket = TicketNumber,
queue = QueueNameBin,
handle_method(#'queue.declare'{queue = QueueNameBin,
passive = true,
nowait = NoWait},
_, State = #ch{ virtual_host = VHostPath }) ->
#ticket{} = lookup_ticket(TicketNumber, #ticket.passive_flag, State),
QueueName = rabbit_misc:r(VHostPath, queue, QueueNameBin),
Q = rabbit_amqqueue:with_or_die(QueueName, fun (Q) -> Q end),
return_queue_declare_ok(State, NoWait, Q);
handle_method(#'queue.delete'{ticket = TicketNumber,
queue = QueueNameBin,
handle_method(#'queue.delete'{queue = QueueNameBin,
if_unused = IfUnused,
if_empty = IfEmpty,
nowait = NoWait
},
_, State) ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
check_ticket(TicketNumber, #ticket.active_flag, QueueName, State),
case rabbit_amqqueue:with_or_die(
QueueName,
fun (Q) -> rabbit_amqqueue:delete(Q, IfUnused, IfEmpty) end) of
@ -640,8 +568,7 @@ handle_method(#'queue.delete'{ticket = TicketNumber,
message_count = PurgedMessageCount})
end;
handle_method(#'queue.bind'{ticket = TicketNumber,
queue = QueueNameBin,
handle_method(#'queue.bind'{queue = QueueNameBin,
exchange = ExchangeNameBin,
routing_key = RoutingKey,
nowait = NoWait,
@ -652,14 +579,13 @@ handle_method(#'queue.bind'{ticket = TicketNumber,
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
ActualRoutingKey = expand_routing_key_shortcut(QueueNameBin, RoutingKey,
State),
check_ticket(TicketNumber, #ticket.active_flag, QueueName, State),
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
case rabbit_amqqueue:add_binding(QueueName, ExchangeName,
ActualRoutingKey, Arguments) of
{error, queue_not_found} ->
{error, queue_not_found} ->
rabbit_misc:protocol_error(
not_found, "no ~s", [rabbit_misc:rs(QueueName)]);
{error, exchange_not_found} ->
{error, exchange_not_found} ->
rabbit_misc:protocol_error(
not_found, "no ~s", [rabbit_misc:rs(ExchangeName)]);
{error, durability_settings_incompatible} ->
@ -670,12 +596,10 @@ handle_method(#'queue.bind'{ticket = TicketNumber,
return_ok(State, NoWait, #'queue.bind_ok'{})
end;
handle_method(#'queue.purge'{ticket = TicketNumber,
queue = QueueNameBin,
handle_method(#'queue.purge'{queue = QueueNameBin,
nowait = NoWait},
_, State) ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
check_ticket(TicketNumber, #ticket.read_flag, QueueName, State),
{ok, PurgedMessageCount} = rabbit_amqqueue:with_or_die(
QueueName,
fun (Q) -> rabbit_amqqueue:purge(Q) end),

74
src/rabbit_control.erl
View File

@ -89,17 +89,6 @@ Available commands:
list_user_vhosts <UserName>
list_vhost_users <VHostPath>
add_realm <VHostPath> <RealmName>
delete_realm <VHostPath> <RealmName>
list_realms <VHostPath>
set_permissions <UserName> <VHostPath> <RealmName> [<Permission> ...]
Permissions management. The available permissions are 'passive',
'active', 'write' and 'read', corresponding to the permissions
referred to in AMQP's \"access.request\" message, or 'all' as an
abbreviation for all defined permission flags.
list_permissions <UserName> <VHostPath>
<node> should be the name of the master node of the RabbitMQ cluster. It
defaults to the node named \"rabbit\" on the local host. On a host named
\"server.example.com\", the master node will usually be rabbit@server (unless
@ -187,68 +176,7 @@ action(list_user_vhosts, Node, Args = [_Username]) ->
action(list_vhost_users, Node, Args = [_VHostPath]) ->
io:format("Listing users for vhosts ~p...", Args),
display_list(call(Node, {rabbit_access_control, list_vhost_users, Args}));
action(add_realm, Node, [VHostPath, RealmName]) ->
io:format("Adding realm ~p to vhost ~p ...", [RealmName, VHostPath]),
rpc_call(Node, rabbit_realm, add_realm,
[realm_rsrc(VHostPath, RealmName)]);
action(delete_realm, Node, [VHostPath, RealmName]) ->
io:format("Deleting realm ~p from vhost ~p ...", [RealmName, VHostPath]),
rpc_call(Node, rabbit_realm, delete_realm,
[realm_rsrc(VHostPath, RealmName)]);
action(list_realms, Node, Args = [_VHostPath]) ->
io:format("Listing realms for vhost ~p ...", Args),
display_list(call(Node, {rabbit_realm, list_vhost_realms, Args}));
action(set_permissions, Node,
[Username, VHostPath, RealmName | Permissions]) ->
io:format("Setting permissions for user ~p, vhost ~p, realm ~p ...",
[Username, VHostPath, RealmName]),
CheckedPermissions = check_permissions(Permissions),
Ticket = #ticket{
realm_name = realm_rsrc(VHostPath, RealmName),
passive_flag = lists:member(passive, CheckedPermissions),
active_flag = lists:member(active, CheckedPermissions),
write_flag = lists:member(write, CheckedPermissions),
read_flag = lists:member(read, CheckedPermissions)},
rpc_call(Node, rabbit_access_control, map_user_realm,
[list_to_binary(Username), Ticket]);
action(list_permissions, Node, Args = [_Username, _VHostPath]) ->
io:format("Listing permissions for user ~p in vhost ~p ...", Args),
Perms = call(Node, {rabbit_access_control, list_user_realms, Args}),
if is_list(Perms) ->
lists:foreach(
fun ({RealmName, Pattern}) ->
io:format("~n~s: ~p",
[binary_to_list(RealmName),
rabbit_misc:permission_list(Pattern)])
end,
lists:sort(Perms)),
io:nl(),
ok;
true -> Perms
end.
check_permissions([]) -> [];
check_permissions(["all" | R]) ->
[passive, active, write, read | check_permissions(R)];
check_permissions([P | R]) when (P == "passive") or
(P == "active") or
(P == "write") or
(P == "read") ->
[list_to_atom(P) | check_permissions(R)];
check_permissions([P | _R]) ->
io:format("~nError: invalid permission flag ~p~n", [P]),
usage().
realm_rsrc(VHostPath, RealmName) ->
rabbit_misc:r(list_to_binary(VHostPath),
realm,
list_to_binary(RealmName)).
display_list(call(Node, {rabbit_access_control, list_vhost_users, Args})).
display_list(L) when is_list(L) ->
lists:foreach(fun (I) ->

5
src/rabbit_error_logger.erl
View File

@ -34,10 +34,7 @@
init([DefaultVHost]) ->
#exchange{} = rabbit_exchange:declare(
#resource{virtual_host = DefaultVHost,
kind = realm,
name = <<"/admin">>},
?LOG_EXCH_NAME,
rabbit_misc:r(DefaultVHost, exchange, ?LOG_EXCH_NAME),
topic, true, false, []),
{ok, #resource{virtual_host = DefaultVHost,
kind = exchange,

38
src/rabbit_exchange.erl
View File

@ -28,7 +28,7 @@
-include("rabbit.hrl").
-include("rabbit_framing.hrl").
-export([recover/0, declare/6, lookup/1, lookup_or_die/1,
-export([recover/0, declare/5, lookup/1, lookup_or_die/1,
list_vhost_exchanges/1, list_exchange_bindings/1,
simple_publish/6, simple_publish/3,
route/2]).
@ -50,21 +50,21 @@
not_found() | {'error', 'unroutable' | 'not_delivered'}).
-spec(recover/0 :: () -> 'ok').
-spec(declare/6 :: (realm_name(), name(), exchange_type(), bool(), bool(),
-spec(declare/5 :: (exchange_name(), exchange_type(), bool(), bool(),
amqp_table()) -> exchange()).
-spec(check_type/1 :: (binary()) -> atom()).
-spec(assert_type/2 :: (exchange(), atom()) -> 'ok').
-spec(assert_type/2 :: (exchange(), atom()) -> 'ok').
-spec(lookup/1 :: (exchange_name()) -> {'ok', exchange()} | not_found()).
-spec(lookup_or_die/1 :: (exchange_name()) -> exchange()).
-spec(list_vhost_exchanges/1 :: (vhost()) -> [exchange()]).
-spec(list_exchange_bindings/1 :: (exchange_name()) ->
-spec(list_exchange_bindings/1 :: (exchange_name()) ->
[{queue_name(), routing_key(), amqp_table()}]).
-spec(simple_publish/6 ::
(bool(), bool(), exchange_name(), routing_key(), binary(), binary()) ->
publish_res()).
-spec(simple_publish/3 :: (bool(), bool(), message()) -> publish_res()).
-spec(route/2 :: (exchange(), routing_key()) -> [pid()]).
-spec(add_binding/2 :: (binding_spec(), amqqueue()) ->
-spec(add_binding/2 :: (binding_spec(), amqqueue()) ->
'ok' | not_found() |
{'error', 'durability_settings_incompatible'}).
-spec(delete_binding/2 :: (binding_spec(), amqqueue()) ->
@ -90,23 +90,21 @@ recover_durable_exchanges() ->
end, ok, durable_exchanges)
end).
declare(RealmName, NameBin, Type, Durable, AutoDelete, Args) ->
XName = rabbit_misc:r(RealmName, exchange, NameBin),
Exchange = #exchange{name = XName,
declare(ExchangeName, Type, Durable, AutoDelete, Args) ->
Exchange = #exchange{name = ExchangeName,
type = Type,
durable = Durable,
auto_delete = AutoDelete,
arguments = Args},
rabbit_misc:execute_mnesia_transaction(
fun () ->
case mnesia:wread({exchange, XName}) of
case mnesia:wread({exchange, ExchangeName}) of
[] -> ok = mnesia:write(Exchange),
if Durable ->
ok = mnesia:write(
durable_exchanges, Exchange, write);
true -> ok
end,
ok = rabbit_realm:add(RealmName, XName),
Exchange;
[ExistingX] -> ExistingX
end
@ -147,15 +145,14 @@ list_vhost_exchanges(VHostPath) ->
list_exchange_bindings(Name) ->
[{QueueName, RoutingKey, Arguments} ||
#binding{handlers = Handlers} <- bindings_for_exchange(Name),
#handler{binding_spec = #binding_spec{routing_key = RoutingKey,
arguments = Arguments},
queue = QueueName} <- Handlers].
#binding{handlers = Handlers} <- bindings_for_exchange(Name),
#handler{binding_spec = #binding_spec{routing_key = RoutingKey,
arguments = Arguments},
queue = QueueName} <- Handlers].
bindings_for_exchange(Name) ->
qlc:e(qlc:q([B ||
B = #binding{key = K} <- mnesia:table(binding),
element(1, K) == Name])).
qlc:e(qlc:q([B || B = #binding{key = K} <- mnesia:table(binding),
element(1, K) == Name])).
empty_handlers() ->
[].
@ -187,7 +184,7 @@ simple_publish(Mandatory, Immediate,
%% return the list of qpids to which a message with a given routing
%% key, sent to a particular exchange, should be delivered.
%%
%%
%% The function ensures that a qpid appears in the return list exactly
%% as many times as a message should be delivered to it. With the
%% current exchange types that is at most once.
@ -197,7 +194,7 @@ route(#exchange{name = Name, type = topic}, RoutingKey) ->
mnesia:activity(
async_dirty,
fun () ->
qlc:e(qlc:q([handler_qpids(H) ||
qlc:e(qlc:q([handler_qpids(H) ||
#binding{key = {Name1, PatternKey},
handlers = H}
<- mnesia:table(binding),
@ -375,6 +372,5 @@ do_internal_delete(ExchangeName, Bindings) ->
ok = mnesia:delete({binding, K})
end, Bindings),
ok = mnesia:delete({durable_exchanges, ExchangeName}),
ok = mnesia:delete({exchange, ExchangeName}),
ok = rabbit_realm:delete_from_all(ExchangeName)
ok = mnesia:delete({exchange, ExchangeName})
end.

76
src/rabbit_misc.erl
View File

@ -29,14 +29,12 @@
-export([method_record_type/1, polite_pause/0, polite_pause/1]).
-export([die/1, frame_error/2, protocol_error/3, protocol_error/4]).
-export([strict_ticket_checking/0]).
-export([get_config/1, get_config/2, set_config/2]).
-export([dirty_read/1]).
-export([r/3, r/2, rs/1]).
-export([permission_list/1]).
-export([enable_cover/0, report_cover/0]).
-export([with_exit_handler/2]).
-export([with_user/2, with_vhost/2, with_realm/2, with_user_and_vhost/3]).
-export([with_user/2, with_vhost/2, with_user_and_vhost/3]).
-export([execute_mnesia_transaction/1]).
-export([ensure_ok/2]).
-export([localnode/1, tcp_name/3]).
@ -64,32 +62,28 @@
(atom() | amqp_error(), string(), [any()]) -> no_return()).
-spec(protocol_error/4 ::
(atom() | amqp_error(), string(), [any()], atom()) -> no_return()).
-spec(strict_ticket_checking/0 :: () -> bool()).
-spec(get_config/1 :: (atom()) -> {'ok', any()} | not_found()).
-spec(get_config/2 :: (atom(), A) -> A).
-spec(set_config/2 :: (atom(), any()) -> 'ok').
-spec(dirty_read/1 :: ({atom(), any()}) -> {'ok', any()} | not_found()).
-spec(r/3 :: (realm_name() | vhost(), K, name()) ->
r(K) when is_subtype(K, atom())).
-spec(r/3 :: (vhost(), K, name()) -> r(K) when is_subtype(K, atom())).
-spec(r/2 :: (vhost(), K) -> #resource{virtual_host :: vhost(),
kind :: K,
name :: '_'}
when is_subtype(K, atom())).
-spec(rs/1 :: (r(atom())) -> string()).
-spec(permission_list/1 :: (ticket()) -> [permission()]).
-spec(rs/1 :: (r(atom())) -> string()).
-spec(enable_cover/0 :: () -> 'ok' | {'error', any()}).
-spec(report_cover/0 :: () -> 'ok').
-spec(with_exit_handler/2 :: (thunk(A), thunk(A)) -> A).
-spec(with_user/2 :: (username(), thunk(A)) -> A).
-spec(with_exit_handler/2 :: (thunk(A), thunk(A)) -> A).
-spec(with_user/2 :: (username(), thunk(A)) -> A).
-spec(with_vhost/2 :: (vhost(), thunk(A)) -> A).
-spec(with_realm/2 :: (realm_name(), thunk(A)) -> A).
-spec(with_user_and_vhost/3 :: (username(), vhost(), thunk(A)) -> A).
-spec(with_user_and_vhost/3 :: (username(), vhost(), thunk(A)) -> A).
-spec(execute_mnesia_transaction/1 :: (thunk(A)) -> A).
-spec(ensure_ok/2 :: ('ok' | {'error', any()}, atom()) -> 'ok').
-spec(ensure_ok/2 :: ('ok' | {'error', any()}, atom()) -> 'ok').
-spec(localnode/1 :: (atom()) -> node()).
-spec(tcp_name/3 :: (atom(), ip_address(), ip_port()) -> atom()).
-spec(tcp_name/3 :: (atom(), ip_address(), ip_port()) -> atom()).
-spec(intersperse/2 :: (A, [A]) -> [A]).
-spec(upmap/2 :: (fun ((A) -> B), [A]) -> [B]).
-spec(upmap/2 :: (fun ((A) -> B), [A]) -> [B]).
-spec(map_in_order/2 :: (fun ((A) -> B), [A]) -> [B]).
-spec(guid/0 :: () -> guid()).
-spec(string_guid/1 :: (any()) -> string()).
@ -128,24 +122,6 @@ protocol_error(Error, Explanation, Params, Method) ->
CompleteExplanation = lists:flatten(io_lib:format(Explanation, Params)),
exit({amqp, Error, CompleteExplanation, Method}).
boolean_config_param(Name, TrueValue, FalseValue, DefaultValue) ->
ActualValue = get_config(Name, DefaultValue),
if
ActualValue == TrueValue ->
true;
ActualValue == FalseValue ->
false;
true ->
rabbit_log:error(
"Bad setting for config param '~w': ~p~n" ++
"legal values are '~w', '~w'; using default value '~w'",
[Name, ActualValue, TrueValue, FalseValue, DefaultValue]),
DefaultValue == TrueValue
end.
strict_ticket_checking() ->
boolean_config_param(strict_ticket_checking, enabled, disabled, disabled).
get_config(Key) ->
case dirty_read({rabbit_config, Key}) of
{ok, {rabbit_config, Key, V}} -> {ok, V};
@ -180,19 +156,6 @@ rs(#resource{virtual_host = VHostPath, kind = Kind, name = Name}) ->
lists:flatten(io_lib:format("~s '~s' in vhost '~s'",
[Kind, Name, VHostPath])).
permission_list(Ticket = #ticket{}) ->
lists:foldr(fun ({Field, Label}, L) ->
case element(Field, Ticket) of
true -> [Label | L];
false -> L
end
end,
[],
[{#ticket.passive_flag, passive},
{#ticket.active_flag, active},
{#ticket.write_flag, write},
{#ticket.read_flag, read}]).
enable_cover() ->
case cover:compile_beam_directory("ebin") of
{error,Reason} -> {error,Reason};
@ -251,32 +214,13 @@ with_user(Username, Thunk) ->
with_vhost(VHostPath, Thunk) ->
fun () ->
case mnesia:read({vhost, VHostPath}) of
[] ->
[] ->
mnesia:abort({no_such_vhost, VHostPath});
[_V] ->
Thunk()
end
end.
with_realm(Name = #resource{virtual_host = VHostPath, kind = realm},
Thunk) ->
fun () ->
case mnesia:read({realm, Name}) of
[] ->
mnesia:abort({no_such_realm, Name});
[_R] ->
case mnesia:match_object(
#vhost_realm{virtual_host = VHostPath,
realm = Name}) of
[] ->
%% This should never happen
mnesia:abort({no_such_realm, Name});
[_VR] ->
Thunk()
end
end
end.
with_user_and_vhost(Username, VHostPath, Thunk) ->
with_user(Username, with_vhost(VHostPath, Thunk)).

23
src/rabbit_mnesia.erl
View File

@ -102,29 +102,6 @@ table_definitions() ->
{index, [virtual_host]}]},
{vhost, [{disc_copies, [node()]},
{attributes, record_info(fields, vhost)}]},
{vhost_realm, [{type, bag},
{disc_copies, [node()]},
{attributes, record_info(fields, vhost_realm)},
{index, [realm]}]},
{realm, [{disc_copies, [node()]},
{attributes, record_info(fields, realm)}]},
{realm_exchange, [{disc_copies, [node()]},
{record_name, realm_resource},
{attributes, record_info(fields, realm_resource)}]},
{realm_queue, [{disc_copies, [node()]},
{record_name, realm_resource},
{attributes, record_info(fields, realm_resource)}]},
{user_realm, [{type, bag},
{disc_copies, [node()]},
{attributes, record_info(fields, user_realm)},
{index, [realm]}]},
{exclusive_realm_visitor,
[{record_name, realm_visitor},
{attributes, record_info(fields, realm_visitor)},
{index, [pid]}]},
{realm_visitor, [{type, bag},
{attributes, record_info(fields, realm_visitor)},
{index, [pid]}]},
{rabbit_config, [{disc_copies, [node()]}]},
{listener, [{type, bag},
{attributes, record_info(fields, listener)}]},

1
src/rabbit_node_monitor.erl
View File

@ -60,7 +60,6 @@ handle_info({nodedown, Node}, State) ->
%% lots of nodes. We really only need to execute this code on
%% *one* node, rather than all of them.
ok = rabbit_networking:on_node_down(Node),
ok = rabbit_realm:on_node_down(Node),
ok = rabbit_amqqueue:on_node_down(Node),
{noreply, State};
handle_info(_Info, State) ->

302
src/rabbit_realm.erl
View File

@ -1,302 +0,0 @@
%% The contents of this file are subject to the Mozilla Public License
%% Version 1.1 (the "License"); you may not use this file except in
%% compliance with the License. You may obtain a copy of the License at
%% http://www.mozilla.org/MPL/
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
%% License for the specific language governing rights and limitations
%% under the License.
%%
%% The Original Code is RabbitMQ.
%%
%% The Initial Developers of the Original Code are LShift Ltd.,
%% Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd.
%%
%% Portions created by LShift Ltd., Cohesive Financial Technologies
%% LLC., and Rabbit Technologies Ltd. are Copyright (C) 2007-2008
%% LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit
%% Technologies Ltd.;
%%
%% All Rights Reserved.
%%
%% Contributor(s): ______________________________________.
%%
-module(rabbit_realm).
-export([recover/0]).
-export([add_realm/1, delete_realm/1, list_vhost_realms/1]).
-export([add/2, delete/2, check/2, delete_from_all/1]).
-export([access_request/3, enter_realm/3, leave_realms/1]).
-export([on_node_down/1]).
-include("rabbit.hrl").
-include_lib("stdlib/include/qlc.hrl").
%%----------------------------------------------------------------------------
-ifdef(use_specs).
-type(e_or_q() :: 'exchange' | 'queue').
-spec(recover/0 :: () -> 'ok').
-spec(add_realm/1 :: (realm_name()) -> 'ok').
-spec(delete_realm/1 :: (realm_name()) -> 'ok').
-spec(list_vhost_realms/1 :: (vhost()) -> [name()]).
-spec(add/2 :: (realm_name(), r(e_or_q())) -> 'ok').
-spec(delete/2 :: (realm_name(), r(e_or_q())) -> 'ok').
-spec(check/2 :: (realm_name(), r(e_or_q())) -> bool() | not_found()).
-spec(delete_from_all/1 :: (r(e_or_q())) -> 'ok').
-spec(access_request/3 :: (username(), bool(), ticket()) ->
'ok' | not_found() | {'error', 'bad_realm_path' |
'access_refused' |
'resource_locked'}).
-spec(enter_realm/3 :: (realm_name(), bool(), pid()) ->
'ok' | {'error', 'resource_locked'}).
-spec(leave_realms/1 :: (pid()) -> 'ok').
-spec(on_node_down/1 :: (node()) -> 'ok').
-endif.
%%--------------------------------------------------------------------
recover() ->
%% preens resource lists, limiting them to currently-extant resources
rabbit_misc:execute_mnesia_transaction(fun preen_realms/0).
add_realm(Name = #resource{virtual_host = VHostPath, kind = realm}) ->
rabbit_misc:execute_mnesia_transaction(
rabbit_misc:with_vhost(
VHostPath,
fun () ->
case mnesia:read({realm, Name}) of
[] ->
NewRealm = #realm{name = Name},
ok = mnesia:write(NewRealm),
ok = mnesia:write(
#vhost_realm{virtual_host = VHostPath,
realm = Name}),
ok;
[_R] ->
mnesia:abort({realm_already_exists, Name})
end
end)).
delete_realm(Name = #resource{virtual_host = VHostPath, kind = realm}) ->
rabbit_misc:execute_mnesia_transaction(
rabbit_misc:with_vhost(
VHostPath,
rabbit_misc:with_realm(
Name,
fun () ->
ok = mnesia:delete({realm, Name}),
ok = mnesia:delete_object(
#vhost_realm{virtual_host = VHostPath,
realm = Name}),
lists:foreach(fun mnesia:delete_object/1,
mnesia:index_read(user_realm, Name,
#user_realm.realm)),
ok
end))).
list_vhost_realms(VHostPath) ->
[Name ||
#vhost_realm{realm = #resource{name = Name}} <-
%% TODO: use dirty ops instead
rabbit_misc:execute_mnesia_transaction(
rabbit_misc:with_vhost(
VHostPath,
fun () -> mnesia:read({vhost_realm, VHostPath}) end))].
add(Realm = #resource{kind = realm}, Resource = #resource{}) ->
manage_link(fun mnesia:write/3, Realm, Resource).
delete(Realm = #resource{kind = realm}, Resource = #resource{}) ->
manage_link(fun mnesia:delete_object/3, Realm, Resource).
% This links or unlinks a resource to a realm
manage_link(Action, Realm = #resource{kind = realm, name = RealmName},
R = #resource{name = Name}) ->
rabbit_misc:execute_mnesia_transaction(
fun () ->
case mnesia:read({realm, Realm}) of
[] -> mnesia:abort(not_found);
[_] -> Action(realm_table_for_resource(R),
#realm_resource{realm = RealmName,
resource = Name},
write)
end
end).
realm_table_for_resource(#resource{kind = exchange}) -> realm_exchange;
realm_table_for_resource(#resource{kind = queue}) -> realm_queue.
parent_table_for_resource(#resource{kind = exchange}) -> exchange;
parent_table_for_resource(#resource{kind = queue}) -> amqqueue.
check(#resource{kind = realm, name = Realm}, R = #resource{name = Name}) ->
case mnesia:dirty_match_object(realm_table_for_resource(R),
#realm_resource{realm = Realm,
resource = Name}) of
[] -> false;
_ -> true
end.
% Requires a mnesia transaction.
delete_from_all(R = #resource{name = Name}) ->
mnesia:delete_object(realm_table_for_resource(R),
#realm_resource{realm = '_', resource = Name},
write).
access_request(Username, Exclusive, Ticket = #ticket{realm_name = RealmName})
when is_binary(Username) ->
%% FIXME: We should do this all in a single tx. Otherwise we may
%% a) get weird answers, b) create inconsistencies in the db
%% (e.g. realm_visitor records referring to non-existing realms).
case check_and_lookup(RealmName) of
{error, Reason} ->
{error, Reason};
{ok, _Realm} ->
{ok, U} = rabbit_access_control:lookup_user(Username),
case rabbit_access_control:lookup_realm_access(U, RealmName) of
none ->
{error, access_refused};
TicketPattern ->
case match_ticket(TicketPattern, Ticket) of
no_match ->
{error, access_refused};
match ->
enter_realm(RealmName, Exclusive, self())
end
end
end.
enter_realm(Name = #resource{kind = realm}, IsExclusive, Pid) ->
RealmVisitor = #realm_visitor{realm = Name, pid = Pid},
rabbit_misc:execute_mnesia_transaction(
fun () ->
case mnesia:read({exclusive_realm_visitor, Name}) of
[] when IsExclusive ->
ok = mnesia:delete_object(RealmVisitor),
%% TODO: find a more efficient way of checking
%% for "no machting results" that doesn't
%% involve retrieving all the records
case mnesia:read({realm_visitor, Name}) of
[] ->
mnesia:write(
exclusive_realm_visitor, RealmVisitor, write),
ok;
[_|_] ->
{error, resource_locked}
end;
[] ->
ok = mnesia:write(RealmVisitor),
ok;
[RealmVisitor] when IsExclusive -> ok;
[RealmVisitor] ->
ok = mnesia:delete({exclusive_realm_visitor, Name}),
ok = mnesia:write(RealmVisitor),
ok;
[_] ->
{error, resource_locked}
end
end).
leave_realms(Pid) ->
rabbit_misc:execute_mnesia_transaction(
fun () ->
case mnesia:index_read(exclusive_realm_visitor, Pid,
#realm_visitor.pid) of
[] -> ok;
[R] ->
ok = mnesia:delete_object(
exclusive_realm_visitor, R, write)
end,
lists:foreach(fun mnesia:delete_object/1,
mnesia:index_read(realm_visitor, Pid,
#realm_visitor.pid)),
ok
end).
on_node_down(Node) ->
rabbit_misc:execute_mnesia_transaction(
fun () ->
lists:foreach(
fun (T) -> ok = remove_visitors(Node, T) end,
[exclusive_realm_visitor, realm_visitor]),
ok
end).
%%--------------------------------------------------------------------
%% This iterates through the realm_exchange and realm_queue link tables
%% and deletes rows that have no underlying exchange or queue record.
preen_realms() ->
lists:foreach(fun preen_realm/1, [exchange, queue]),
ok.
preen_realm(Kind) ->
R = #resource{kind = Kind},
Table = realm_table_for_resource(R),
Cursor = qlc:cursor(
qlc:q([L#realm_resource.resource ||
L <- mnesia:table(Table)])),
preen_next(Cursor, Table, parent_table_for_resource(R)),
qlc:delete_cursor(Cursor).
preen_next(Cursor, Table, ParentTable) ->
case qlc:next_answers(Cursor, 1) of
[] -> ok;
[Name] ->
case mnesia:read({ParentTable, Name}) of
[] -> mnesia:delete_object(
Table,
#realm_resource{realm = '_', resource = Name},
write);
_ -> ok
end,
preen_next(Cursor, Table, ParentTable)
end.
check_and_lookup(RealmName = #resource{kind = realm,
name = <<"/data", _/binary>>}) ->
lookup(RealmName);
check_and_lookup(RealmName = #resource{kind = realm,
name = <<"/admin", _/binary>>}) ->
lookup(RealmName);
check_and_lookup(_) ->
{error, bad_realm_path}.
lookup(Name = #resource{kind = realm}) ->
rabbit_misc:dirty_read({realm, Name}).
match_ticket(#ticket{passive_flag = PP,
active_flag = PA,
write_flag = PW,
read_flag = PR},
#ticket{passive_flag = TP,
active_flag = TA,
write_flag = TW,
read_flag = TR}) ->
if
%% Matches if either we're not requesting passive access, or
%% passive access is permitted, and ...
(not(TP) orelse PP) andalso
(not(TA) orelse PA) andalso
(not(TW) orelse PW) andalso
(not(TR) orelse PR) ->
match;
true ->
no_match
end.
remove_visitors(Node, T) ->
qlc:fold(
fun (R, Acc) ->
ok = mnesia:delete_object(T, R, write),
Acc
end,
ok,
qlc:q([R || R = #realm_visitor{pid = Pid} <- mnesia:table(T),
node(Pid) == Node])).

51
src/rabbit_tests.erl
View File

@ -284,31 +284,12 @@ test_user_management() ->
control_action(unmap_user_vhost, ["foo", "/"]),
{error, {no_such_user, _}} =
control_action(list_user_vhosts, ["foo"]),
{error, {no_such_user, _}} =
control_action(set_permissions, ["foo", "/", "/data"]),
{error, {no_such_user, _}} =
control_action(list_permissions, ["foo", "/"]),
{error, {no_such_vhost, _}} =
control_action(map_user_vhost, ["guest", "/testhost"]),
{error, {no_such_vhost, _}} =
control_action(unmap_user_vhost, ["guest", "/testhost"]),
{error, {no_such_vhost, _}} =
control_action(list_vhost_users, ["/testhost"]),
{error, {no_such_vhost, _}} =
control_action(set_permissions, ["guest", "/testhost", "/data"]),
{error, {no_such_vhost, _}} =
control_action(list_permissions, ["guest", "/testhost"]),
{error, {no_such_vhost, _}} =
control_action(add_realm, ["/testhost", "/data/test"]),
{error, {no_such_vhost, _}} =
control_action(delete_realm, ["/testhost", "/data/test"]),
{error, {no_such_vhost, _}} =
control_action(list_realms, ["/testhost"]),
{error, {no_such_realm, _}} =
control_action(set_permissions, ["guest", "/", "/data/test"]),
{error, {no_such_realm, _}} =
control_action(delete_realm, ["/", "/data/test"]),
%% user creation
ok = control_action(add_user, ["foo", "bar"]),
{error, {user_already_exists, _}} =
@ -327,32 +308,6 @@ test_user_management() ->
ok = control_action(map_user_vhost, ["foo", "/testhost"]),
ok = control_action(list_user_vhosts, ["foo"]),
%% realm creation
ok = control_action(add_realm, ["/testhost", "/data/test"]),
{error, {realm_already_exists, _}} =
control_action(add_realm, ["/testhost", "/data/test"]),
ok = control_action(list_realms, ["/testhost"]),
%% user permissions
ok = control_action(set_permissions,
["foo", "/testhost", "/data/test",
"passive", "active", "write", "read"]),
ok = control_action(list_permissions, ["foo", "/testhost"]),
ok = control_action(set_permissions,
["foo", "/testhost", "/data/test", "all"]),
ok = control_action(set_permissions,
["foo", "/testhost", "/data/test"]),
{error, not_mapped_to_vhost} =
control_action(set_permissions,
["guest", "/testhost", "/data/test"]),
{error, not_mapped_to_vhost} =
control_action(list_permissions, ["guest", "/testhost"]),
%% realm deletion
ok = control_action(delete_realm, ["/testhost", "/data/test"]),
{error, {no_such_realm, _}} =
control_action(delete_realm, ["/testhost", "/data/test"]),
%% user/vhost unmapping
ok = control_action(unmap_user_vhost, ["foo", "/testhost"]),
ok = control_action(unmap_user_vhost, ["foo", "/testhost"]),
@ -364,13 +319,7 @@ test_user_management() ->
%% deleting a populated vhost
ok = control_action(add_vhost, ["/testhost"]),
ok = control_action(add_realm, ["/testhost", "/data/test"]),
ok = control_action(map_user_vhost, ["foo", "/testhost"]),
ok = control_action(set_permissions,
["foo", "/testhost", "/data/test", "all"]),
_ = rabbit_amqqueue:declare(
rabbit_misc:r(<<"/testhost">>, realm, <<"/data/test">>),
<<"bar">>, true, false, []),
ok = control_action(delete_vhost, ["/testhost"]),
%% user deletion

131
src/rabbit_ticket.erl
View File

@ -1,131 +0,0 @@
%% The contents of this file are subject to the Mozilla Public License
%% Version 1.1 (the "License"); you may not use this file except in
%% compliance with the License. You may obtain a copy of the License at
%% http://www.mozilla.org/MPL/
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
%% License for the specific language governing rights and limitations
%% under the License.
%%
%% The Original Code is RabbitMQ.
%%
%% The Initial Developers of the Original Code are LShift Ltd.,
%% Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd.
%%
%% Portions created by LShift Ltd., Cohesive Financial Technologies
%% LLC., and Rabbit Technologies Ltd. are Copyright (C) 2007-2008
%% LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit
%% Technologies Ltd.;
%%
%% All Rights Reserved.
%%
%% Contributor(s): ______________________________________.
%%
-module(rabbit_ticket).
-include("rabbit.hrl").
-export([record_ticket/2, lookup_ticket/4, check_ticket/4]).
-import(application).
%%----------------------------------------------------------------------------
-ifdef(use_specs).
-type(ticket_number() :: non_neg_integer()).
%% we'd like to write #ticket.passive_flag | #ticket.active_flag | ...
%% but dialyzer doesn't support that.
-type(ticket_field() :: 3..6).
-spec(record_ticket/2 :: (ticket_number(), ticket()) -> 'ok').
-spec(lookup_ticket/4 ::
(ticket_number(), ticket_field(), username(), vhost()) ->
ticket()).
-spec(check_ticket/4 ::
(ticket_number(), ticket_field(), r('exchange' | 'queue'), username()) ->
'ok').
-endif.
%%----------------------------------------------------------------------------
record_ticket(TicketNumber, Ticket) ->
put({ticket, TicketNumber}, Ticket),
ok.
lookup_ticket(TicketNumber, FieldIndex, Username, VHostPath) ->
case get({ticket, TicketNumber}) of
undefined ->
%% Spec: "The server MUST isolate access tickets per
%% channel and treat an attempt by a client to mix these
%% as a connection exception."
rabbit_log:warning("Attempt by client to use invalid ticket ~p~n", [TicketNumber]),
maybe_relax_checks(TicketNumber, Username, VHostPath);
Ticket = #ticket{} ->
case element(FieldIndex, Ticket) of
false -> rabbit_misc:protocol_error(
access_refused,
"ticket ~w has insufficient permissions",
[TicketNumber]);
true -> Ticket
end
end.
maybe_relax_checks(TicketNumber, Username, VHostPath) ->
case rabbit_misc:strict_ticket_checking() of
true ->
rabbit_misc:protocol_error(
access_refused, "invalid ticket ~w", [TicketNumber]);
false ->
rabbit_log:warning("Lax ticket check mode: fabricating full ticket ~p for user ~p, vhost ~p~n",
[TicketNumber, Username, VHostPath]),
Ticket = rabbit_access_control:full_ticket(
rabbit_misc:r(VHostPath, realm, <<"/data">>)),
case rabbit_realm:access_request(Username, false, Ticket) of
ok -> record_ticket(TicketNumber, Ticket),
Ticket;
{error, Reason} ->
rabbit_misc:protocol_error(
Reason,
"fabrication of ticket ~w for user '~s' in vhost '~s' failed",
[TicketNumber, Username, VHostPath])
end
end.
check_ticket(TicketNumber, FieldIndex,
Name = #resource{virtual_host = VHostPath}, Username) ->
#ticket{realm_name = RealmName} =
lookup_ticket(TicketNumber, FieldIndex, Username, VHostPath),
case resource_in_realm(RealmName, Name) of
false ->
case rabbit_misc:strict_ticket_checking() of
true ->
rabbit_misc:protocol_error(
access_refused,
"insufficient permissions in ticket ~w to access ~s in ~s",
[TicketNumber, rabbit_misc:rs(Name),
rabbit_misc:rs(RealmName)]);
false ->
rabbit_log:warning("Lax ticket check mode: ignoring cross-realm access for ticket ~p~n", [TicketNumber]),
ok
end;
true ->
ok
end.
resource_in_realm(RealmName, ResourceName = #resource{kind = Kind}) ->
CacheKey = {resource_cache, RealmName, Kind},
case get(CacheKey) of
Name when Name == ResourceName ->
true;
_ ->
case rabbit_realm:check(RealmName, ResourceName) of
true ->
put(CacheKey, ResourceName),
true;
_ ->
false
end
end.